danvics/telegram-files
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

🚑️ fix: Add maintenance program to support the fix for missing album message caption.

Browse source
This commit is contained in:
jarvis2f 2025-02-17 11:45:38 +08:00
parent 0f66477ba2
commit e03bb0fb7f
19 changed files with 484 additions and 156 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
Dockerfile
View file

@ -77,7 +77,9 @@ RUN npm install -g pm2 && \
apt-get clean && \
rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/* ./libs.zip && \
touch /run/nginx.pid && \
chown -R 1000:1000 /app /etc/nginx /var/lib/nginx /var/log/nginx /run/nginx.pid
chown -R 1000:1000 /app /etc/nginx /var/lib/nginx /var/log/nginx /run/nginx.pid && \
echo '#!/bin/sh\njava -Djava.library.path=/app/tdlib -cp /app/api.jar telegram.files.Maintain "$@"' > /usr/bin/tfm && \
chmod +x /usr/bin/tfm
COPY --from=runtime-builder --chown=1000:1000 /custom-jre/jre /jre
COPY --from=api-builder --chown=1000:1000 /app/api.jar /app/api.jar

21
README.md
View file

@ -222,3 +222,24 @@ manually.~~
Open the `chrome://flags` page, search for `Insecure origins treated as secure`, and add the address of the web page to
the list.
</details>
**Q.** How to use the telegram-files maintenance tool?
**A.** You can use the following command to run the maintenance tool(**before running the command, you should stop telegram-files container**):
<details closed>
<summary>Command</summary>
```shell
docker run --rm \
--entrypoint /bin/bash \
-v $(pwd)/data:/app/data \
-e APP_ROOT=${APP_ROOT:-/app/data} \
-e TELEGRAM_API_ID=${TELEGRAM_API_ID} \
-e TELEGRAM_API_HASH=${TELEGRAM_API_HASH} \
ghcr.io/jarvis2f/telegram-files:latest tfm ${Maintenance Command}
```
**Maintenance Command:**
- `album-caption`: Fixed issue with missing caption for album messages before `0.1.15`.
</details>

11
api/src/main/java/telegram/files/AutoDownloadVerticle.java
View file

@ -130,7 +130,7 @@ public class AutoDownloadVerticle extends AbstractVerticle {
}
Tuple2<String, List<String>> rule = handleRule(auto);
TelegramVerticle telegramVerticle = this.getTelegramVerticle(auto.telegramId);
TelegramVerticle telegramVerticle = TelegramVerticles.getOrElseThrow(auto.telegramId);
TdApi.SearchChatMessages searchChatMessages = new TdApi.SearchChatMessages();
searchChatMessages.query = rule.v1;
searchChatMessages.chatId = auto.chatId;
@ -229,7 +229,7 @@ public class AutoDownloadVerticle extends AbstractVerticle {
return;
}
log.debug("Download start! TelegramId: %d size: %d".formatted(telegramId, messages.size()));
TelegramVerticle telegramVerticle = this.getTelegramVerticle(telegramId);
TelegramVerticle telegramVerticle = TelegramVerticles.getOrElseThrow(telegramId);
int surplusSize = getSurplusSize(telegramId);
if (surplusSize <= 0) {
return;
@ -251,11 +251,6 @@ public class AutoDownloadVerticle extends AbstractVerticle {
log.debug("Remaining download messages: %d".formatted(messages.size()));
}
private TelegramVerticle getTelegramVerticle(long telegramId) {
return HttpVerticle.getTelegramVerticle(telegramId)
.orElseThrow(() -> new IllegalArgumentException("Telegram verticle not found: %s".formatted(telegramId)));
}
private void onNewMessage(JsonObject jsonObject) {
long telegramId = jsonObject.getLong("telegramId");
long chatId = jsonObject.getLong("chatId");
@ -263,7 +258,7 @@ public class AutoDownloadVerticle extends AbstractVerticle {
autoRecords.items.stream()
.filter(item -> item.telegramId == telegramId && item.chatId == chatId)
.findFirst()
.flatMap(item -> HttpVerticle.getTelegramVerticle(telegramId))
.flatMap(item -> TelegramVerticles.get(telegramId))
.ifPresent(telegramVerticle -> {
if (telegramVerticle.authorized) {
telegramVerticle.client.execute(new TdApi.GetMessage(chatId, messageId))

4
api/src/main/java/telegram/files/AutoRecordsHolder.java
View file

@ -35,7 +35,7 @@ public class AutoRecordsHolder {
if (settingAutoRecords == null) {
return;
}
settingAutoRecords.items.forEach(item -> HttpVerticle.getTelegramVerticle(item.telegramId)
settingAutoRecords.items.forEach(item -> TelegramVerticles.get(item.telegramId)
.ifPresentOrElse(telegramVerticle -> {
if (telegramVerticle.authorized) {
autoRecords.add(item);
@ -52,7 +52,7 @@ public class AutoRecordsHolder {
for (SettingAutoRecords.Item item : records.items) {
if (!autoRecords.exists(item.telegramId, item.chatId)) {
// new enabled
HttpVerticle.getTelegramVerticle(item.telegramId)
TelegramVerticles.get(item.telegramId)
.ifPresentOrElse(telegramVerticle -> {
if (telegramVerticle.authorized) {
autoRecords.add(item);

5
api/src/main/java/telegram/files/Config.java
View file

@ -7,6 +7,8 @@ import cn.hutool.core.util.StrUtil;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import cn.hutool.log.dialect.jdk.JdkLog;
import io.vertx.core.DeploymentOptions;
import io.vertx.core.ThreadingModel;
import java.io.File;
import java.io.IOException;
@ -29,6 +31,9 @@ public class Config {
public static final int TELEGRAM_LOG_LEVEL = Convert.toInt(System.getenv("TELEGRAM_LOG_LEVEL"), 0);
public static final DeploymentOptions VIRTUAL_THREAD_DEPLOYMENT_OPTIONS = new DeploymentOptions()
.setThreadingModel(ThreadingModel.VIRTUAL_THREAD);
private static Level logLevel;
static {

6
api/src/main/java/telegram/files/EventEnum.java
View file

@ -31,6 +31,12 @@ public enum EventEnum {
* @see telegram.files.EventPayload
*/
TELEGRAM_EVENT,
/**
* suffix = null <br>
* body = JSONObject with "success", "message"
*/
MAINTAIN,
;
public String address() {

187
api/src/main/java/telegram/files/HttpVerticle.java
View file

@ -3,7 +3,6 @@ package telegram.files;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.core.util.URLUtil;
import cn.hutool.log.Log;
@ -14,7 +13,6 @@ import io.vertx.core.http.CookieSameSite;
import io.vertx.core.http.HttpMethod;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.http.HttpServerResponse;
import io.vertx.core.impl.NoStackTraceException;
import io.vertx.core.json.Json;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
@ -31,9 +29,7 @@ import org.drinkless.tdlib.TdApi;
import telegram.files.repository.SettingAutoRecords;
import telegram.files.repository.SettingKey;
import telegram.files.repository.SettingRecord;
import telegram.files.repository.TelegramRecord;
import java.io.File;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
@ -45,8 +41,6 @@ public class HttpVerticle extends AbstractVerticle {
// session id -> ws handler id
private static final Map<String, String> clients = new ConcurrentHashMap<>();
private static final List<TelegramVerticle> telegramVerticles = new ArrayList<>();
// session id -> telegram verticle
private final Map<String, TelegramVerticle> sessionTelegramVerticles = new ConcurrentHashMap<>();
@ -184,52 +178,16 @@ public class HttpVerticle extends AbstractVerticle {
}
public Future<Void> initTelegramVerticles() {
return DataVerticle.telegramRepository.getAll()
.compose(telegramRecords -> {
List<String> verifiedPath = telegramRecords.stream().map(TelegramRecord::rootPath).toList();
File telegramRoot = FileUtil.file(Config.TELEGRAM_ROOT);
List<String> uncertifiedPaths = FileUtil.loopFiles(telegramRoot, 1, null)
.stream()
.filter(f -> f.isDirectory() && !verifiedPath.contains(f.getAbsolutePath()))
.map(File::getAbsolutePath)
.toList();
List<Future<String>> futures = new ArrayList<>();
for (TelegramRecord telegramRecord : telegramRecords) {
TelegramVerticle telegramVerticle = new TelegramVerticle(telegramRecord);
if (!telegramVerticle.check()) {
continue;
}
telegramVerticles.add(telegramVerticle);
futures.add(vertx.deployVerticle(telegramVerticle));
}
if (CollUtil.isNotEmpty(uncertifiedPaths)) {
for (String uncertifiedPath : uncertifiedPaths) {
TelegramVerticle telegramVerticle = new TelegramVerticle(uncertifiedPath);
if (!telegramVerticle.check()) {
continue;
}
telegramVerticles.add(telegramVerticle);
futures.add(vertx.deployVerticle(telegramVerticle));
}
}
return Future.all(futures);
})
.onSuccess(r -> log.info("Successfully deployed %d telegram verticles".formatted(r.size())))
.onFailure(err -> log.error("Failed to deploy telegram verticles: %s".formatted(err.getMessage())))
.mapEmpty();
return TelegramVerticles.initTelegramVerticles(vertx);
}
public Future<Void> initAutoDownloadVerticle() {
return vertx.deployVerticle(new AutoDownloadVerticle(autoRecordsHolder), new DeploymentOptions()
.setThreadingModel(ThreadingModel.VIRTUAL_THREAD)
)
return vertx.deployVerticle(new AutoDownloadVerticle(autoRecordsHolder), Config.VIRTUAL_THREAD_DEPLOYMENT_OPTIONS)
.mapEmpty();
}
public Future<Void> initTransferVerticle() {
return vertx.deployVerticle(new TransferVerticle(autoRecordsHolder), new DeploymentOptions()
.setThreadingModel(ThreadingModel.VIRTUAL_THREAD)
)
return vertx.deployVerticle(new TransferVerticle(autoRecordsHolder), Config.VIRTUAL_THREAD_DEPLOYMENT_OPTIONS)
.mapEmpty();
}
@ -350,7 +308,7 @@ public class HttpVerticle extends AbstractVerticle {
TelegramVerticle newTelegramVerticle = new TelegramVerticle(DataVerticle.telegramRepository.getRootPath());
newTelegramVerticle.setProxy(proxyName);
sessionTelegramVerticles.put(sessionId, newTelegramVerticle);
telegramVerticles.add(newTelegramVerticle);
TelegramVerticles.add(newTelegramVerticle);
vertx.deployVerticle(newTelegramVerticle)
.onSuccess(id -> ctx.json(new JsonObject()
.put("id", newTelegramVerticle.getId())
@ -360,20 +318,13 @@ public class HttpVerticle extends AbstractVerticle {
}
private void handleTelegramDelete(RoutingContext ctx) {
String telegramId = ctx.pathParam("telegramId");
if (StrUtil.isBlank(telegramId)) {
ctx.fail(400);
TelegramVerticle telegramVerticle = getTelegramVerticleByPath(ctx);
if (telegramVerticle == null) {
return;
}
Optional<TelegramVerticle> telegramVerticleOptional = getTelegramVerticle(telegramId);
if (telegramVerticleOptional.isEmpty()) {
ctx.fail(404);
return;
}
TelegramVerticle telegramVerticle = telegramVerticleOptional.get();
telegramVerticle.close(true)
.onSuccess(r -> {
telegramVerticles.remove(telegramVerticle);
TelegramVerticles.remove(telegramVerticle);
sessionTelegramVerticles.entrySet().removeIf(e -> e.getValue().equals(telegramVerticle));
ctx.end();
});
@ -381,7 +332,7 @@ public class HttpVerticle extends AbstractVerticle {
private void handleTelegrams(RoutingContext ctx) {
Boolean authorized = Convert.toBool(ctx.request().getParam("authorized"));
Future.all(telegramVerticles.stream()
Future.all(TelegramVerticles.getAll().stream()
.filter(c -> authorized == null || c.authorized == authorized)
.map(TelegramVerticle::getTelegramAccount)
.toList()
@ -392,29 +343,25 @@ public class HttpVerticle extends AbstractVerticle {
}
private void handleTelegramChats(RoutingContext ctx) {
String telegramId = ctx.pathParam("telegramId");
if (StrUtil.isBlank(telegramId)) {
ctx.fail(400);
TelegramVerticle telegramVerticle = getTelegramVerticleByPath(ctx);
if (telegramVerticle == null) {
return;
}
String query = ctx.request().getParam("query");
String chatId = ctx.request().getParam("chatId");
String archived = ctx.request().getParam("archived");
getTelegramVerticle(telegramId)
.ifPresentOrElse(telegramVerticle ->
telegramVerticle.getChats(Convert.toLong(chatId), query, Convert.toBool(archived, false))
.onSuccess(ctx::json)
.onFailure(ctx::fail), () -> ctx.fail(404));
telegramVerticle.getChats(Convert.toLong(chatId), query, Convert.toBool(archived, false))
.onSuccess(ctx::json)
.onFailure(ctx::fail);
}
private void handleTelegramFiles(RoutingContext ctx) {
String telegramId = ctx.pathParam("telegramId");
if (StrUtil.isBlank(telegramId)) {
ctx.fail(400);
TelegramVerticle telegramVerticle = getTelegramVerticleByPath(ctx);
if (telegramVerticle == null) {
return;
}
String chatIdStr = ctx.pathParam("chatId");
if (StrUtil.isBlank(chatIdStr)) {
String chatId = ctx.pathParam("chatId");
if (StrUtil.isBlank(chatId)) {
ctx.fail(400);
return;
}
@ -422,19 +369,14 @@ public class HttpVerticle extends AbstractVerticle {
ctx.request().params().forEach(filter::put);
filter.put("search", URLUtil.decode(filter.get("search")));
long chatId = Convert.toLong(chatIdStr);
getTelegramVerticle(telegramId)
.ifPresentOrElse(telegramVerticle ->
telegramVerticle.getChatFiles(chatId, filter)
.onSuccess(ctx::json)
.onFailure(ctx::fail),
() -> ctx.fail(404));
telegramVerticle.getChatFiles(Convert.toLong(chatId), filter)
.onSuccess(ctx::json)
.onFailure(ctx::fail);
}
private void handleTelegramFilesCount(RoutingContext ctx) {
String telegramId = ctx.pathParam("telegramId");
if (StrUtil.isBlank(telegramId)) {
ctx.fail(400);
TelegramVerticle telegramVerticle = getTelegramVerticleByPath(ctx);
if (telegramVerticle == null) {
return;
}
String chatIdStr = ctx.pathParam("chatId");
@ -443,26 +385,16 @@ public class HttpVerticle extends AbstractVerticle {
return;
}
long chatId = Convert.toLong(chatIdStr);
getTelegramVerticle(telegramId)
.ifPresentOrElse(telegramVerticle ->
telegramVerticle.getChatFilesCount(chatId)
.onSuccess(ctx::json)
.onFailure(ctx::fail),
() -> ctx.fail(404));
telegramVerticle.getChatFilesCount(chatId)
.onSuccess(ctx::json)
.onFailure(ctx::fail);
}
private void handleTelegramDownloadStatistics(RoutingContext ctx) {
String telegramId = ctx.pathParam("telegramId");
if (StrUtil.isBlank(telegramId)) {
ctx.fail(400);
TelegramVerticle telegramVerticle = getTelegramVerticleByPath(ctx);
if (telegramVerticle == null) {
return;
}
Optional<TelegramVerticle> telegramVerticleOptional = getTelegramVerticle(telegramId);
if (telegramVerticleOptional.isEmpty()) {
ctx.fail(404);
return;
}
TelegramVerticle telegramVerticle = telegramVerticleOptional.get();
String type = ctx.request().getParam("type");
String timeRange = ctx.request().getParam("timeRange");
@ -487,7 +419,7 @@ public class HttpVerticle extends AbstractVerticle {
sessionTelegramVerticles.remove(sessionId);
return true;
}
Optional<TelegramVerticle> optionalTelegramVerticle = getTelegramVerticle(telegramId);
Optional<TelegramVerticle> optionalTelegramVerticle = TelegramVerticles.get(telegramId);
if (optionalTelegramVerticle.isEmpty()) {
return false;
}
@ -497,7 +429,7 @@ public class HttpVerticle extends AbstractVerticle {
private void handleTelegramToggleProxy(RoutingContext ctx) {
String telegramId = ctx.request().getParam("telegramId");
getTelegramVerticle(telegramId)
TelegramVerticles.get(telegramId)
.ifPresentOrElse(telegramVerticle ->
telegramVerticle.toggleProxy(ctx.body().asJsonObject())
.onSuccess(r -> ctx.json(JsonObject.of("proxy", r)))
@ -510,7 +442,7 @@ public class HttpVerticle extends AbstractVerticle {
ctx.fail(400);
return;
}
getTelegramVerticle(telegramId)
TelegramVerticles.get(telegramId)
.ifPresentOrElse(telegramVerticle ->
telegramVerticle.ping()
.onSuccess(r -> ctx.json(JsonObject.of("ping", r)))
@ -524,7 +456,7 @@ public class HttpVerticle extends AbstractVerticle {
ctx.fail(400);
return;
}
getTelegramVerticle(telegramId)
TelegramVerticles.get(telegramId)
.ifPresentOrElse(telegramVerticle ->
telegramVerticle.client.execute(new TdApi.TestNetwork())
.onComplete(r ->
@ -549,7 +481,7 @@ public class HttpVerticle extends AbstractVerticle {
ctx.fail(400);
return;
}
TelegramVerticle telegramVerticle = getTelegramVerticle(ctx);
TelegramVerticle telegramVerticle = getTelegramVerticleBySession(ctx);
if (telegramVerticle == null) {
return;
}
@ -560,12 +492,10 @@ public class HttpVerticle extends AbstractVerticle {
}
private void handleFilePreview(RoutingContext ctx) {
Optional<TelegramVerticle> telegramVerticleOptional = getTelegramVerticle(ctx.pathParam("telegramId"));
if (telegramVerticleOptional.isEmpty()) {
ctx.fail(404);
TelegramVerticle telegramVerticle = getTelegramVerticleByPath(ctx);
if (telegramVerticle == null) {
return;
}
TelegramVerticle telegramVerticle = telegramVerticleOptional.get();
String uniqueId = ctx.pathParam("uniqueId");
if (StrUtil.isBlank(uniqueId)) {
ctx.fail(404);
@ -585,7 +515,7 @@ public class HttpVerticle extends AbstractVerticle {
}
private void handleFileStartDownload(RoutingContext ctx) {
TelegramVerticle telegramVerticle = getTelegramVerticleThrow(ctx.pathParam("telegramId"));
TelegramVerticle telegramVerticle = TelegramVerticles.getOrElseThrow(ctx.pathParam("telegramId"));
JsonObject jsonObject = ctx.body().asJsonObject();
Long chatId = jsonObject.getLong("chatId");
@ -602,7 +532,7 @@ public class HttpVerticle extends AbstractVerticle {
}
private void handleFileStartDownloadMultiple(RoutingContext ctx) {
TelegramVerticle telegramVerticle = getTelegramVerticleThrow(ctx.pathParam("telegramId"));
TelegramVerticle telegramVerticle = TelegramVerticles.getOrElseThrow(ctx.pathParam("telegramId"));
JsonObject jsonObject = ctx.body().asJsonObject();
Long chatId = jsonObject.getLong("chatId");
@ -629,7 +559,7 @@ public class HttpVerticle extends AbstractVerticle {
}
private void handleFileCancelDownload(RoutingContext ctx) {
TelegramVerticle telegramVerticle = getTelegramVerticleThrow(ctx.pathParam("telegramId"));
TelegramVerticle telegramVerticle = TelegramVerticles.getOrElseThrow(ctx.pathParam("telegramId"));
JsonObject jsonObject = ctx.body().asJsonObject();
Integer fileId = jsonObject.getInteger("fileId");
@ -644,7 +574,7 @@ public class HttpVerticle extends AbstractVerticle {
}
private void handleFileTogglePauseDownload(RoutingContext ctx) {
TelegramVerticle telegramVerticle = getTelegramVerticleThrow(ctx.pathParam("telegramId"));
TelegramVerticle telegramVerticle = TelegramVerticles.getOrElseThrow(ctx.pathParam("telegramId"));
JsonObject jsonObject = ctx.body().asJsonObject();
Integer fileId = jsonObject.getInteger("fileId");
@ -660,7 +590,7 @@ public class HttpVerticle extends AbstractVerticle {
}
private void handleFileRemove(RoutingContext ctx) {
TelegramVerticle telegramVerticle = getTelegramVerticleThrow(ctx.pathParam("telegramId"));
TelegramVerticle telegramVerticle = TelegramVerticles.getOrElseThrow(ctx.pathParam("telegramId"));
if (telegramVerticle == null) {
return;
}
@ -678,7 +608,7 @@ public class HttpVerticle extends AbstractVerticle {
}
private void handleAutoDownload(RoutingContext ctx) {
TelegramVerticle telegramVerticle = getTelegramVerticleThrow(ctx.pathParam("telegramId"));
TelegramVerticle telegramVerticle = TelegramVerticles.getOrElseThrow(ctx.pathParam("telegramId"));
String chatId = ctx.request().getParam("chatId");
if (StrUtil.isBlank(chatId)) {
@ -691,28 +621,7 @@ public class HttpVerticle extends AbstractVerticle {
.onFailure(ctx::fail);
}
public static Optional<TelegramVerticle> getTelegramVerticle(String telegramId) {
Object id = NumberUtil.isNumber(telegramId) ? Convert.toLong(telegramId) : telegramId;
return telegramVerticles.stream()
.filter(t -> Objects.equals(t.getId(), id))
.findFirst();
}
public static TelegramVerticle getTelegramVerticleThrow(String telegramId) {
Object id = NumberUtil.isNumber(telegramId) ? Convert.toLong(telegramId) : telegramId;
return telegramVerticles.stream()
.filter(t -> Objects.equals(t.getId(), id))
.findFirst()
.orElseThrow(() -> new NoStackTraceException("Telegram account not found!"));
}
public static Optional<TelegramVerticle> getTelegramVerticle(long telegramId) {
return telegramVerticles.stream()
.filter(t -> t.telegramRecord != null && t.telegramRecord.id() == telegramId)
.findFirst();
}
private TelegramVerticle getTelegramVerticle(RoutingContext ctx) {
private TelegramVerticle getTelegramVerticleBySession(RoutingContext ctx) {
String sessionId = ctx.session().id();
TelegramVerticle telegramVerticle = sessionTelegramVerticles.get(sessionId);
if (telegramVerticle == null) {
@ -722,4 +631,18 @@ public class HttpVerticle extends AbstractVerticle {
}
return telegramVerticle;
}
private TelegramVerticle getTelegramVerticleByPath(RoutingContext ctx) {
String telegramId = ctx.pathParam("telegramId");
if (StrUtil.isBlank(telegramId)) {
ctx.fail(400);
return null;
}
Optional<TelegramVerticle> telegramVerticleOptional = TelegramVerticles.get(telegramId);
if (telegramVerticleOptional.isEmpty()) {
ctx.fail(404);
return null;
}
return telegramVerticleOptional.get();
}
}

63
api/src/main/java/telegram/files/Maintain.java Normal file
View file

@ -0,0 +1,63 @@
package telegram.files;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import io.vertx.core.Vertx;
import telegram.files.maintains.AlbumCaptionMaintainVerticle;
import telegram.files.maintains.MaintainVerticle;
public class Maintain {
static {
LogFactory.setCurrentLogFactory(new Config.JDKLogFactory());
}
private static final Log log = LogFactory.get();
private static final Vertx vertx = Vertx.vertx();
public static void main(String[] args) {
if (ArrayUtil.isEmpty(args)) {
System.out.println("Missing maintain name");
System.out.println("Usage: java -cp api.jar telegram.files.Maintain <maintain-name>");
System.out.println("Maintain names:");
System.out.println(" album-caption");
System.exit(1);
}
String maintainName = args[0];
try {
MaintainVerticle maintainVerticle = null;
switch (maintainName) {
case "album-caption" -> {
maintainVerticle = new AlbumCaptionMaintainVerticle();
MessyUtils.await(vertx.deployVerticle(maintainVerticle, Config.VIRTUAL_THREAD_DEPLOYMENT_OPTIONS)
.onFailure(err -> {
log.error("Failed to deploy album caption maintain verticle", err);
System.exit(1);
}));
}
default -> {
System.out.println("Unknown maintain name: " + maintainName);
System.exit(1);
}
}
final MaintainVerticle finalMaintainVerticle = maintainVerticle;
vertx.eventBus().consumer(EventEnum.MAINTAIN.address(), message ->
vertx.undeploy(finalMaintainVerticle.deploymentID())
.onSuccess(v -> {
log.trace("Undeploy maintain verticle success");
System.exit(0);
})
.onFailure(err -> {
log.error("Failed to undeploy maintain verticle", err);
System.exit(1);
}));
} catch (Exception e) {
log.error("Failed to maintain", e);
System.exit(1);
}
}
}

2
api/src/main/java/telegram/files/Start.java
View file

@ -13,7 +13,7 @@ public class Start {
private static final Log log = LogFactory.get();
public static final String VERSION = "0.1.14";
public static final String VERSION = "0.1.15";
private static final CountDownLatch shutdownLatch = new CountDownLatch(1);

4
api/src/main/java/telegram/files/TdApiHelp.java
View file

@ -273,6 +273,7 @@ public class TdApiHelp {
telegramId,
message.chatId,
message.id,
message.mediaAlbumId,
message.date,
message.hasSensitiveContent,
file.size == 0 ? file.expectedSize : file.size,
@ -342,6 +343,7 @@ public class TdApiHelp {
telegramId,
message.chatId,
message.id,
message.mediaAlbumId,
message.date,
message.hasSensitiveContent,
file.size == 0 ? file.expectedSize : file.size,
@ -399,6 +401,7 @@ public class TdApiHelp {
telegramId,
message.chatId,
message.id,
message.mediaAlbumId,
message.date,
message.hasSensitiveContent,
file.size == 0 ? file.expectedSize : file.size,
@ -447,6 +450,7 @@ public class TdApiHelp {
telegramId,
message.chatId,
message.id,
message.mediaAlbumId,
message.date,
message.hasSensitiveContent,
file.size == 0 ? file.expectedSize : file.size,

95
api/src/main/java/telegram/files/TelegramVerticles.java Normal file
View file

@ -0,0 +1,95 @@
package telegram.files;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.impl.NoStackTraceException;
import telegram.files.repository.TelegramRecord;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
public class TelegramVerticles {
private static final Log log = LogFactory.get();
private static final List<TelegramVerticle> telegramVerticles = new ArrayList<>();
public static Future<Void> initTelegramVerticles(Vertx vertx) {
return DataVerticle.telegramRepository.getAll()
.compose(telegramRecords -> {
List<String> verifiedPath = telegramRecords.stream().map(TelegramRecord::rootPath).toList();
File telegramRoot = FileUtil.file(Config.TELEGRAM_ROOT);
List<String> uncertifiedPaths = FileUtil.loopFiles(telegramRoot, 1, null)
.stream()
.filter(f -> f.isDirectory() && !verifiedPath.contains(f.getAbsolutePath()))
.map(File::getAbsolutePath)
.toList();
List<Future<String>> futures = new ArrayList<>();
for (TelegramRecord telegramRecord : telegramRecords) {
TelegramVerticle telegramVerticle = new TelegramVerticle(telegramRecord);
if (!telegramVerticle.check()) {
continue;
}
telegramVerticles.add(telegramVerticle);
futures.add(vertx.deployVerticle(telegramVerticle));
}
if (CollUtil.isNotEmpty(uncertifiedPaths)) {
for (String uncertifiedPath : uncertifiedPaths) {
TelegramVerticle telegramVerticle = new TelegramVerticle(uncertifiedPath);
if (!telegramVerticle.check()) {
continue;
}
telegramVerticles.add(telegramVerticle);
futures.add(vertx.deployVerticle(telegramVerticle));
}
}
return Future.all(futures);
})
.onSuccess(r -> log.info("Successfully deployed %d telegram verticles".formatted(r.size())))
.onFailure(err -> log.error("Failed to deploy telegram verticles: %s".formatted(err.getMessage())))
.mapEmpty();
}
public static void add(TelegramVerticle telegramVerticle) {
telegramVerticles.add(telegramVerticle);
}
public static void remove(TelegramVerticle telegramVerticle) {
telegramVerticles.remove(telegramVerticle);
}
public static List<TelegramVerticle> getAll() {
return telegramVerticles;
}
public static Optional<TelegramVerticle> get(String telegramId) {
Object id = NumberUtil.isNumber(telegramId) ? Convert.toLong(telegramId) : telegramId;
return telegramVerticles.stream()
.filter(t -> Objects.equals(t.getId(), id))
.findFirst();
}
public static TelegramVerticle getOrElseThrow(String telegramId) {
return get(telegramId)
.orElseThrow(() -> new NoStackTraceException("Telegram account not found!"));
}
public static Optional<TelegramVerticle> get(long telegramId) {
return telegramVerticles.stream()
.filter(t -> t.telegramRecord != null && t.telegramRecord.id() == telegramId)
.findFirst();
}
public static TelegramVerticle getOrElseThrow(long telegramId) {
return get(telegramId)
.orElseThrow(() -> new NoStackTraceException("Telegram account not found!"));
}
}

141
api/src/main/java/telegram/files/maintains/AlbumCaptionMaintainVerticle.java Normal file
View file

@ -0,0 +1,141 @@
package telegram.files.maintains;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.IterUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.sqlclient.templates.SqlTemplate;
import org.drinkless.tdlib.TdApi;
import telegram.files.Config;
import telegram.files.DataVerticle;
import telegram.files.TelegramVerticle;
import telegram.files.TelegramVerticles;
import telegram.files.repository.FileRecord;
import java.util.List;
import java.util.Optional;
/**
* This verticle is responsible for maintaining the captions of the media files.
*/
public class AlbumCaptionMaintainVerticle extends MaintainVerticle {
private static final Log log = LogFactory.get();
private final DataVerticle dataVerticle = new DataVerticle();
@Override
public void start(Promise<Void> startPromise) {
vertx.deployVerticle(dataVerticle, Config.VIRTUAL_THREAD_DEPLOYMENT_OPTIONS)
.compose(r -> TelegramVerticles.initTelegramVerticles(vertx))
.compose(r -> {
vertx.setTimer(1000, id -> handleAlbumCaption());
return Future.succeededFuture();
})
.onSuccess(id -> startPromise.complete())
.onFailure(startPromise::fail);
}
public void handleAlbumCaption() {
log.info("🔨 Start to handle album caption");
try {
log.trace("🔨 1.Scan all file records and update the media_album_id of file records");
long fromMessageId = 0;
long count = 0;
long page = 1;
while (true) {
log.trace("🔨 Scan page %d, limit 100".formatted(page));
List<FileRecord> rows = Future.await(SqlTemplate.forQuery(DataVerticle.pool, """
SELECT * FROM file_record WHERE media_album_id is null %s ORDER BY message_id desc LIMIT 100
""".formatted(fromMessageId == 0 ? "" : " AND message_id < #{fromMessageId}")
)
.mapTo(FileRecord.ROW_MAPPER)
.execute(MapUtil.of("fromMessageId", fromMessageId))
.map(IterUtil::toList));
if (CollUtil.isEmpty(rows)) {
log.trace("🔨 No more file records found, update finished");
break;
}
for (FileRecord fileRecord : rows) {
if (updateMediaAlbumId(fileRecord)) {
count++;
}
}
fromMessageId = rows.getLast().messageId();
page++;
}
log.info("✅ Updated %d file records with media album id".formatted(count));
log.trace("🔨 2.Update the caption of the media album");
fromMessageId = 0;
count = 0;
page = 1;
while (true) {
log.trace("🔨 Scan page %d, limit 100".formatted(page));
List<FileRecord> rows = Future.await(SqlTemplate.forQuery(DataVerticle.pool, """
SELECT * FROM file_record WHERE media_album_id is not null AND caption != '' %s ORDER BY message_id desc LIMIT 100
""".formatted(fromMessageId == 0 ? "" : " AND message_id < #{fromMessageId}")
)
.mapTo(FileRecord.ROW_MAPPER)
.execute(MapUtil.of("fromMessageId", fromMessageId))
.map(IterUtil::toList));
if (CollUtil.isEmpty(rows)) {
log.trace("🔨 No more file records found, update finished");
break;
}
for (FileRecord fileRecord : rows) {
int updated = Future.await(DataVerticle.fileRepository.updateCaptionByMediaAlbumId(fileRecord.mediaAlbumId(), fileRecord.caption()));
count += updated;
}
fromMessageId = rows.getLast().messageId();
page++;
}
log.info("✅ Updated %d media albums with caption".formatted(count));
log.info("✅ Finished handling album caption");
super.end(true, null);
} catch (Exception e) {
log.error("🔨 Failed to handle album caption", e);
super.end(false, e);
}
}
private boolean updateMediaAlbumId(FileRecord fileRecord) {
try {
Optional<TelegramVerticle> telegramVerticleOptional = TelegramVerticles.get(fileRecord.telegramId());
if (telegramVerticleOptional.isEmpty()) {
log.error("🔨 Telegram verticle not found for telegram id: %d".formatted(fileRecord.telegramId()));
return false;
}
TelegramVerticle telegramVerticle = telegramVerticleOptional.get();
TdApi.Message message = Future.await(telegramVerticle.client.execute(new TdApi.GetMessage(fileRecord.chatId(), fileRecord.messageId())));
if (message != null && message.mediaAlbumId != 0) {
Future.await(SqlTemplate.forUpdate(DataVerticle.pool, """
UPDATE file_record
SET media_album_id = #{mediaAlbumId}
WHERE unique_id = #{uniqueId}
""")
.execute(MapUtil.ofEntries(
MapUtil.entry("uniqueId", fileRecord.uniqueId()),
MapUtil.entry("mediaAlbumId", message.mediaAlbumId)
))
.onFailure(err -> log.error("🔨 Failed to update media album id: %s".formatted(err.getMessage())))
.map(true));
return true;
}
return false;
} catch (Exception e) {
log.error(e, "🔨 Failed to update media album id, unique id: %s".formatted(fileRecord.uniqueId()));
return false;
}
}
}

14
api/src/main/java/telegram/files/maintains/MaintainVerticle.java Normal file
View file

@ -0,0 +1,14 @@
package telegram.files.maintains;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.json.JsonObject;
import telegram.files.EventEnum;
public class MaintainVerticle extends AbstractVerticle {
public void end(boolean success, Throwable cause) {
vertx.eventBus().publish(EventEnum.MAINTAIN.address(),
JsonObject.of("success", success, "message", cause == null ? null : cause.getMessage())
);
}
}

9
api/src/main/java/telegram/files/repository/FileRecord.java
View file

@ -14,6 +14,7 @@ public record FileRecord(int id, //file id will change
long telegramId,
long chatId,
long messageId,
long mediaAlbumId,
int date, // date when the file was uploaded
boolean hasSensitiveContent,
long size, // file size in bytes
@ -46,6 +47,7 @@ public record FileRecord(int id, //file id will change
telegram_id BIGINT,
chat_id BIGINT,
message_id BIGINT,
media_album_id BIGINT,
date INT,
has_sensitive_content BOOLEAN,
size BIGINT,
@ -71,6 +73,9 @@ public record FileRecord(int id, //file id will change
}),
MapUtil.entry(new Version("0.1.12"), new String[]{
"ALTER TABLE file_record ADD COLUMN transfer_status VARCHAR(255) DEFAULT 'idle';",
}),
MapUtil.entry(new Version("0.1.15"), new String[]{
"ALTER TABLE file_record ADD COLUMN media_album_id BIGINT;",
})
));
@ -92,6 +97,7 @@ public record FileRecord(int id, //file id will change
row.getLong("telegram_id"),
row.getLong("chat_id"),
row.getLong("message_id"),
Objects.requireNonNullElse(row.getLong("media_album_id"), 0L),
row.getInteger("date"),
Convert.toBool(row.getInteger("has_sensitive_content")),
row.getLong("size"),
@ -115,6 +121,7 @@ public record FileRecord(int id, //file id will change
MapUtil.entry("telegram_id", r.telegramId()),
MapUtil.entry("chat_id", r.chatId()),
MapUtil.entry("message_id", r.messageId()),
MapUtil.entry("media_album_id", r.mediaAlbumId()),
MapUtil.entry("date", r.date()),
MapUtil.entry("has_sensitive_content", r.hasSensitiveContent()),
MapUtil.entry("size", r.size()),
@ -132,7 +139,7 @@ public record FileRecord(int id, //file id will change
));
public FileRecord withSourceField(int id, long downloadedSize) {
return new FileRecord(id, uniqueId, telegramId, chatId, messageId, date, hasSensitiveContent, size, downloadedSize, type, mimeType, fileName, thumbnail, caption, localPath, downloadStatus, transferStatus, startDate, completionDate);
return new FileRecord(id, uniqueId, telegramId, chatId, messageId, mediaAlbumId, date, hasSensitiveContent, size, downloadedSize, type, mimeType, fileName, thumbnail, caption, localPath, downloadStatus, transferStatus, startDate, completionDate);
}
public boolean isDownloadStatus(DownloadStatus status) {

4
api/src/main/java/telegram/files/repository/FileRepository.java
View file

@ -21,6 +21,8 @@ public interface FileRepository {
Future<FileRecord> getByUniqueId(String uniqueId);
Future<String> getCaptionByMediaAlbumId(long mediaAlbumId);
Future<JsonObject> getDownloadStatistics(long telegramId);
Future<JsonArray> getCompletedRangeStatistics(long id, long startTime, long endTime, int timeRange);
@ -39,5 +41,7 @@ public interface FileRepository {
Future<Void> updateFileId(int fileId, String uniqueId);
Future<Integer> updateCaptionByMediaAlbumId(long mediaAlbumId, String caption);
Future<Void> deleteByUniqueId(String uniqueId);
}

54
api/src/main/java/telegram/files/repository/impl/FileRepositoryImpl.java
View file

@ -13,6 +13,7 @@ import io.vertx.core.Future;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.jdbcclient.JDBCPool;
import io.vertx.sqlclient.SqlResult;
import io.vertx.sqlclient.templates.SqlTemplate;
import org.jooq.lambda.tuple.Tuple;
import org.jooq.lambda.tuple.Tuple3;
@ -38,16 +39,20 @@ public class FileRepositoryImpl implements FileRepository {
public Future<FileRecord> create(FileRecord fileRecord) {
return SqlTemplate
.forUpdate(pool, """
INSERT INTO file_record(id, unique_id, telegram_id, chat_id, message_id, date, has_sensitive_content, size, downloaded_size,
INSERT INTO file_record(id, unique_id, telegram_id, chat_id, message_id, media_album_id, date, has_sensitive_content,
size, downloaded_size,
type, mime_type,
file_name, thumbnail, caption, local_path,
download_status, start_date, transfer_status)
values (#{id}, #{unique_id}, #{telegram_id}, #{chat_id}, #{message_id}, #{date}, #{has_sensitive_content}, #{size}, #{downloaded_size}, #{type},
#{mime_type}, #{file_name}, #{thumbnail}, #{caption}, #{local_path}, #{download_status}, #{start_date}, #{transfer_status})
values (#{id}, #{unique_id}, #{telegram_id}, #{chat_id}, #{message_id}, #{media_album_id}, #{date},
#{has_sensitive_content}, #{size}, #{downloaded_size}, #{type},
#{mime_type}, #{file_name}, #{thumbnail}, #{caption}, #{local_path}, #{download_status}, #{start_date},
#{transfer_status})
""")
.mapFrom(FileRecord.PARAM_MAPPER)
.execute(fileRecord)
.map(r -> fileRecord)
.compose(r -> this.updateCaptionByMediaAlbumId(fileRecord.mediaAlbumId(), fileRecord.caption()).map(r))
.onSuccess(r -> log.trace("Successfully created file record: %s".formatted(fileRecord.id()))
)
.onFailure(
@ -195,6 +200,21 @@ public class FileRepositoryImpl implements FileRepository {
.map(rs -> rs.size() > 0 ? rs.iterator().next() : null);
}
@Override
public Future<String> getCaptionByMediaAlbumId(long mediaAlbumId) {
if (mediaAlbumId <= 0) {
return Future.succeededFuture(null);
}
return SqlTemplate
.forQuery(pool, """
SELECT caption FROM file_record WHERE media_album_id = #{mediaAlbumId} LIMIT 1
""")
.mapTo(row -> row.getString("caption"))
.execute(Map.of("mediaAlbumId", mediaAlbumId))
.map(rs -> rs.size() > 0 ? rs.iterator().next() : null)
.onFailure(err -> log.error("Failed to get caption: %s".formatted(err.getMessage())));
}
@Override
public Future<JsonObject> getDownloadStatistics(long telegramId) {
return SqlTemplate
@ -425,6 +445,34 @@ public class FileRepositoryImpl implements FileRepository {
});
}
@Override
public Future<Integer> updateCaptionByMediaAlbumId(long mediaAlbumId, String caption) {
if (mediaAlbumId <= 0) {
return Future.succeededFuture(0);
}
return Future.<String>future(promise -> {
if (StrUtil.isBlank(caption)) {
this.getCaptionByMediaAlbumId(mediaAlbumId)
.onSuccess(promise::complete);
} else {
promise.complete(caption);
}
}).compose(theCaption -> {
if (StrUtil.isBlank(theCaption)) {
return Future.succeededFuture(0);
}
return SqlTemplate
.forUpdate(pool, """
UPDATE file_record SET caption = #{caption} WHERE media_album_id = #{mediaAlbumId}
""")
.execute(Map.of("mediaAlbumId", mediaAlbumId, "caption", theCaption))
.onFailure(err -> log.error("Failed to update file record: %s".formatted(err.getMessage()))
)
.map(SqlResult::rowCount);
});
}
@Override
public Future<Void> deleteByUniqueId(String uniqueId) {
if (StrUtil.isBlank(uniqueId)) {

8
api/src/test/java/telegram/files/AutoRecordsHolderTest.java
View file

@ -20,7 +20,7 @@ import static org.mockito.Mockito.*;
public class AutoRecordsHolderTest {
private AutoRecordsHolder autoRecordsHolder;
private MockedStatic<HttpVerticle> httpVerticleMockedStatic;
private MockedStatic<TelegramVerticles> telegramVerticlesMockedStatic;
private SettingAutoRecords settingAutoRecords1;
@ -29,7 +29,7 @@ public class AutoRecordsHolderTest {
@BeforeEach
public void setUp() {
autoRecordsHolder = new AutoRecordsHolder();
httpVerticleMockedStatic = mockStatic(HttpVerticle.class);
telegramVerticlesMockedStatic = mockStatic(TelegramVerticles.class);
// Prepare test data
settingAutoRecords1 = new SettingAutoRecords();
@ -41,13 +41,13 @@ public class AutoRecordsHolderTest {
// Mock dependencies
TelegramVerticle mockTelegramVerticle = mock(TelegramVerticle.class);
mockTelegramVerticle.authorized = true;
when(HttpVerticle.getTelegramVerticle(item1.telegramId))
when(TelegramVerticles.get(item1.telegramId))
.thenReturn(Optional.of(mockTelegramVerticle));
}
@AfterEach
public void tearDown() {
httpVerticleMockedStatic.close();
telegramVerticlesMockedStatic.close();
}
@Test

6
api/src/test/java/telegram/files/DataVerticleTest.java
View file

@ -82,7 +82,7 @@ public class DataVerticleTest {
@DisplayName("Test Get file record by primary key")
void getFileRecordByPrimaryKeyTest(Vertx vertx, VertxTestContext testContext) {
FileRecord fileRecord = new FileRecord(
1, "unique_id", 1, 1, 1, 1, false, 1, 0, "type", "mime_type", "file_name", "thumbnail", "caption", "local_path", "download_status", "transfer_status", 0, null
1, "unique_id", 1, 1, 1, 1, 1, false, 1, 0, "type", "mime_type", "file_name", "thumbnail", "caption", "local_path", "download_status", "transfer_status", 0, null
);
DataVerticle.fileRepository.create(fileRecord)
.compose(r -> DataVerticle.fileRepository.getByPrimaryKey(r.id(), r.uniqueId()))
@ -96,7 +96,7 @@ public class DataVerticleTest {
@DisplayName("Test update file download status")
void updateFileDownloadStatusTest(Vertx vertx, VertxTestContext testContext) {
FileRecord fileRecord = new FileRecord(
1, "unique_id", 1, 1, 1, 1, false, 1, 0, "type", "mime_type", "file_name", "thumbnail", "caption", null, FileRecord.DownloadStatus.idle.name(), FileRecord.TransferStatus.idle.name(), 0, null
1, "unique_id", 1, 1, 1, 1, 1, false, 1, 0, "type", "mime_type", "file_name", "thumbnail", "caption", null, FileRecord.DownloadStatus.idle.name(), FileRecord.TransferStatus.idle.name(), 0, null
);
String updateLocalPath = "local_path";
Long completionDate = 1L;
@ -123,7 +123,7 @@ public class DataVerticleTest {
@DisplayName("Test update file transfer status")
void updateFileTransferStatusTest(Vertx vertx, VertxTestContext testContext) {
FileRecord fileRecord = new FileRecord(
1, "unique_id", 1, 1, 1, 1, false, 1, 0, "type", "mime_type", "file_name", "thumbnail", "caption", null, FileRecord.DownloadStatus.idle.name(), FileRecord.TransferStatus.idle.name(), 0, null
1, "unique_id", 1, 1, 1, 1, 1, false, 1, 0, "type", "mime_type", "file_name", "thumbnail", "caption", null, FileRecord.DownloadStatus.idle.name(), FileRecord.TransferStatus.idle.name(), 0, null
);
String updateLocalPath = "local_path";
DataVerticle.fileRepository.create(fileRecord)

2
api/src/test/java/telegram/files/FileDownloadStatusConcurrentTest.java
View file

@ -24,7 +24,7 @@ public class FileDownloadStatusConcurrentTest {
private static ExecutorService executor;
static FileRecord fileRecord = new FileRecord(
1, "unique_id", 1, 1, 1, 1, false, 1, 0, "type", "mime_type", "file_name", "thumbnail", "caption", null,
1, "unique_id", 1, 1, 1, 1, 1, false, 1, 0, "type", "mime_type", "file_name", "thumbnail", "caption", null,
FileRecord.DownloadStatus.idle.name(), FileRecord.TransferStatus.idle.name(), 0, null
);