CERBERUS/cobalt
2
1
Fork 0
mirror of https://github.com/imputnet/cobalt.git synced 2026-10-24 11:02:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix: clean up interrupted Android downloads

Browse source
This commit is contained in:
wolfmanster 2026-08-19 22:07:04 +08:00
parent 9c13ced057
commit 9fe1f89718
4 changed files with 26 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

7
Android/project/app/src/main/kotlin/com/xmedia/archive/data/ArchiveDao.kt
View file

@ -36,8 +36,8 @@ interface ArchiveDao {
@Query("DELETE FROM media WHERE jobId = :jobId")
suspend fun deleteMedia(jobId: String)
@Query("DELETE FROM jobs WHERE status IN ('completed', 'failed', 'canceled')")
suspend fun deleteHistory(): Int
@Query("DELETE FROM jobs WHERE id IN (:ids) AND status IN ('completed', 'failed', 'canceled')")
suspend fun deleteTerminalJobs(ids: List<String>): Int
@Query("DELETE FROM media WHERE jobId NOT IN (SELECT id FROM jobs)")
suspend fun deleteOrphanMedia()
@ -45,6 +45,9 @@ interface ArchiveDao {
@Query("UPDATE jobs SET status = 'queued', progress = 0, error = NULL, updatedAt = :updatedAt WHERE status IN ('resolving', 'downloading')")
suspend fun requeueInterrupted(updatedAt: String): Int
@Query("SELECT * FROM jobs WHERE status IN ('resolving', 'downloading')")
suspend fun interruptedJobs(): List<JobEntity>
@Query("SELECT EXISTS(SELECT 1 FROM jobs WHERE status IN ('queued', 'resolving', 'downloading'))")
suspend fun hasPendingJobs(): Boolean

17
Android/project/app/src/main/kotlin/com/xmedia/archive/repository/ArchiveRepository.kt
View file

@ -52,7 +52,10 @@ class ArchiveRepository(context: Context) {
suspend fun update(job: JobEntity) = dao.upsertJob(job.copy(updatedAt = Instant.now().toString()))
suspend fun requeueInterrupted() = dao.requeueInterrupted(Instant.now().toString())
suspend fun requeueInterrupted(): Int {
dao.interruptedJobs().forEach { job -> removeMedia(job.id) }
return dao.requeueInterrupted(Instant.now().toString())
}
suspend fun hasPendingJobs() = dao.hasPendingJobs()
@ -60,10 +63,11 @@ class ArchiveRepository(context: Context) {
suspend fun clearHistory(): Int {
val terminal = dao.listJobs().filter { it.status in setOf("completed", "failed", "canceled") }
if (terminal.isEmpty()) return 0
terminal.flatMap { dao.mediaForJob(it.id) }.mapNotNull { it.mediaStoreUri }.forEach { uri ->
runCatching { appContext.contentResolver.delete(android.net.Uri.parse(uri), null, null) }
}
val removed = dao.deleteHistory()
val removed = dao.deleteTerminalJobs(terminal.map { it.id })
dao.deleteOrphanMedia()
return removed
}
@ -71,7 +75,7 @@ class ArchiveRepository(context: Context) {
suspend fun retry(id: String): Boolean {
val job = dao.getJob(id) ?: return false
if (job.status != JobStatus.FAILED.name.lowercase() && job.status != JobStatus.CANCELED.name.lowercase()) return false
dao.deleteMedia(id)
removeMedia(id)
update(job.copy(status = JobStatus.QUEUED.name.lowercase(), progress = 0, error = null, completedAt = null))
return true
}
@ -87,6 +91,13 @@ class ArchiveRepository(context: Context) {
suspend fun jobJson(id: String): JSONObject? = dao.getJob(id)?.let { toJson(it, dao.mediaForJob(id)) }
private suspend fun removeMedia(jobId: String) {
dao.mediaForJob(jobId).mapNotNull { it.mediaStoreUri }.forEach { uri ->
runCatching { appContext.contentResolver.delete(android.net.Uri.parse(uri), null, null) }
}
dao.deleteMedia(jobId)
}
private fun toJson(job: JobEntity, media: List<MediaEntity>): JSONObject {
val result = JSONObject()
.put("id", job.id).put("tweetId", job.tweetId).put("sourceUrl", job.sourceUrl)

8
Android/project/app/src/main/kotlin/com/xmedia/archive/resolver/XPostResolver.kt
View file

@ -6,6 +6,7 @@ import okhttp3.Request
import okhttp3.RequestBody.Companion.toRequestBody
import org.json.JSONArray
import org.json.JSONObject
import kotlinx.coroutines.CancellationException
import java.net.URI
import java.time.Instant
import java.util.UUID
@ -45,7 +46,7 @@ class XPostResolver(private val client: OkHttpClient = OkHttpClient()) {
}
fun resolve(tweetId: String, signal: () -> Boolean = { false }): ResolvedPost {
if (signal()) throw InterruptedException("任务已取消")
if (signal()) throw CancellationException("任务已取消")
val syndication = runCatching { requestSyndication(tweetId, signal) }.getOrNull()
if (syndication != null) return syndication
return resolveGraphql(tweetId, signal)
@ -53,12 +54,13 @@ class XPostResolver(private val client: OkHttpClient = OkHttpClient()) {
}
private fun requestSyndication(tweetId: String, signal: () -> Boolean): ResolvedPost {
if (signal()) throw InterruptedException("任务已取消")
if (signal()) throw CancellationException("任务已取消")
val token = base36((tweetId.toDouble() / 1e15) * PI).replace(Regex("(0+|\\.)"), "")
val requestUrl = "https://cdn.syndication.twimg.com/tweet-result?id=$tweetId&token=$token&lang=zh-cn"
val request = Request.Builder().url(requestUrl).header("User-Agent", USER_AGENT).header("Accept", "application/json").build()
client.newCall(request).execute().use { response ->
if (!response.isSuccessful) throw IllegalStateException("Syndication HTTP ${response.code}")
if (signal()) throw CancellationException("任务已取消")
return postFromSyndication(JSONObject(response.body?.string() ?: "{}"), tweetId)
}
}
@ -88,7 +90,7 @@ class XPostResolver(private val client: OkHttpClient = OkHttpClient()) {
private fun resolveGraphql(tweetId: String, signal: () -> Boolean): ResolvedPost? {
var token = guestToken() ?: return null
repeat(2) { attempt ->
if (signal()) throw InterruptedException("任务已取消")
if (signal()) throw CancellationException("任务已取消")
val url = GRAPHQL_URL.toHttpUrl().newBuilder()
.addQueryParameter("variables", JSONObject().put("focalTweetId", tweetId).put("with_rux_injections", false).put("rankingMode", "Relevance").put("includePromotedContent", true).put("withCommunity", true).toString())
.addQueryParameter("features", GRAPHQL_FEATURES)

2
Android/project/app/src/main/kotlin/com/xmedia/archive/service/DownloadForegroundService.kt
View file

@ -112,6 +112,7 @@ class DownloadForegroundService : Service() {
repository.update(job)
val resolved = resolver.resolve(job.tweetId) { canceledJobs.contains(job.id) }
coroutineContext.ensureActive()
if (canceledJobs.contains(job.id)) throw CancellationException("任务已取消")
val media = resolved.media.map {
MediaEntity(it.id, job.id, it.kind, it.filename, sourceUrl = it.sourceUrl, position = it.position)
}
@ -163,6 +164,7 @@ class DownloadForegroundService : Service() {
}
val uri = contentResolver.insert(collection, values) ?: throw IllegalStateException("无法创建媒体文件")
try {
ArchiveDatabase.get(this).dao().upsertMedia(listOf(item.copy(contentType = contentType, mediaStoreUri = uri.toString())))
contentResolver.openOutputStream(uri)?.use { output ->
body.byteStream().use { input ->
val buffer = ByteArray(DEFAULT_BUFFER_SIZE)