nanxun/imagefind
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ci): satisfy backend lint checks

Browse Source
This commit is contained in:
nanxun
2026-08-12 20:58:33 +08:00
parent a0a185b96c
commit 9b8880a3da
2 changed files with 73 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/imagefind/api.py
+59 -50
View File
@@ -64,9 +64,7 @@ async def _background_api_with_fallback(function, fallback, /, *args, **kwargs):
task = asyncio.create_task(_background_api(function, *args, **kwargs)) task = asyncio.create_task(_background_api(function, *args, **kwargs))
try: try:
return await asyncio.wait_for( return await asyncio.wait_for(asyncio.shield(task), timeout=API_READ_DEADLINE_SECONDS)
asyncio.shield(task), timeout=API_READ_DEADLINE_SECONDS
)
except TimeoutError: except TimeoutError:
# asyncio.to_thread cannot stop a running database call. Let it refresh # asyncio.to_thread cannot stop a running database call. Let it refresh
# the service cache, consume any eventual exception, and serve the last # the service cache, consume any eventual exception, and serve the last
@@ -627,9 +625,7 @@ def _set_session(
*, *,
client_key: str = "local", client_key: str = "local",
) -> dict: ) -> dict:
token, csrf, expires = app.auth.login( token, csrf, expires = app.auth.login(password, remember_device=remember_device, client_key=client_key)
password, remember_device=remember_device, client_key=client_key
)
return _set_cookie( return _set_cookie(
response, response,
app, app,
@@ -1390,13 +1386,11 @@ def home_feed(
for row in tag_rows for row in tag_rows
] ]
unorganized_items = ( unorganized_items = (
videos(app, current_user, organized=False, sort="added", limit=item_limit) videos(app, current_user, organized=False, sort="added", limit=item_limit) if unorganized_count else []
if unorganized_count
else []
) )
hide_unorganized = unorganized_count <= item_limit and { hide_unorganized = unorganized_count <= item_limit and {item["id"] for item in unorganized_items} == {
item["id"] for item in unorganized_items item["id"] for item in recent
} == {item["id"] for item in recent} }
actor_items = sorted( actor_items = sorted(
(item for item in list_actors(app, current_user) if int(item["video_count"] or 0) > 0), (item for item in list_actors(app, current_user) if int(item["video_count"] or 0) > 0),
key=lambda item: (-int(item["video_count"] or 0), str(item["name"]).casefold()), key=lambda item: (-int(item["video_count"] or 0), str(item["name"]).casefold()),
@@ -1788,9 +1782,7 @@ async def jobs(
if page is not None: if page is not None:
if page < 1 or page_size < 1 or page_size > 50: if page < 1 or page_size < 1 or page_size > 50:
raise HTTPException(422, "page 必须大于 0，page_size 必须在 1–50 之间") raise HTTPException(422, "page 必须大于 0，page_size 必须在 1–50 之间")
return await _background_api( return await _background_api(app.jobs.paginate, page, page_size, lane=lane, status=status)
app.jobs.paginate, page, page_size, lane=lane, status=status
)
return await _background_api(app.jobs.list, limit, lane=lane, status=status) return await _background_api(app.jobs.list, limit, lane=lane, status=status)
@@ -1905,27 +1897,43 @@ async def search(
if isinstance(rebuild, dict) and rebuild.get("status") in {"running", "failed"}: if isinstance(rebuild, dict) and rebuild.get("status") in {"running", "failed"}:
if rebuild.get("status") == "running": if rebuild.get("status") == "running":
with app.db.read() as conn: with app.db.read() as conn:
remaining = int(conn.execute( remaining = int(
"SELECT count(*) FROM videos WHERE available=1 AND audio_index_revision<?", conn.execute(
(SPEECH_INDEX_REVISION,), "SELECT count(*) FROM videos WHERE available=1 AND audio_index_revision<?",
).fetchone()[0]) (SPEECH_INDEX_REVISION,),
active = int(conn.execute( ).fetchone()[0]
"SELECT count(*) FROM jobs WHERE kind='transcribe_audio' AND status IN ('queued','running')" )
).fetchone()[0]) active = int(
failed = int(conn.execute( conn.execute(
"SELECT count(*) FROM jobs WHERE kind='transcribe_audio' AND status='failed' AND finished_at>=?", "SELECT count(*) FROM jobs WHERE kind='transcribe_audio' AND status IN ('queued','running')"
(rebuild.get("started_at") or "",), ).fetchone()[0]
).fetchone()[0]) )
failed = int(
conn.execute(
"SELECT count(*) FROM jobs WHERE kind='transcribe_audio' AND status='failed' AND finished_at>=?", # noqa: E501
(rebuild.get("started_at") or "",),
).fetchone()[0]
)
if active == 0: if active == 0:
status = "completed" if remaining == 0 else "failed" status = "completed" if remaining == 0 else "failed"
rebuild = {**rebuild, "status": status, "remaining": remaining, "failed_jobs": failed, "finished_at": utcnow()} rebuild = {
**rebuild,
"status": status,
"remaining": remaining,
"failed_jobs": failed,
"finished_at": utcnow(),
}
app.db.set_setting("audio_rebuild", rebuild) app.db.set_setting("audio_rebuild", rebuild)
if rebuild.get("status") == "completed": if rebuild.get("status") == "completed":
rebuild = {} rebuild = {}
else: else:
requested = set(body.recognition_types or []) requested = set(body.recognition_types or [])
if "audio" in requested: if "audio" in requested:
message = "音频索引全库重建存在失败任务，修复前音频搜索保持暂停" if rebuild.get("status") == "failed" else "音频索引正在全库重建，音频搜索暂时不可用" message = (
"音频索引全库重建存在失败任务，修复前音频搜索保持暂停"
if rebuild.get("status") == "failed"
else "音频索引正在全库重建，音频搜索暂时不可用"
)
raise HTTPException(409, message) raise HTTPException(409, message)
if body.recognition_types is None: if body.recognition_types is None:
body.recognition_types = ["visual", "ocr", "person", "subtitle", "metadata"] body.recognition_types = ["visual", "ocr", "person", "subtitle", "metadata"]
@@ -1969,8 +1977,7 @@ async def search(
cjk = sum("\u3400" <= character <= "\u9fff" for character in compact) cjk = sum("\u3400" <= character <= "\u9fff" for character in compact)
if compact and ((cjk == len(compact) and cjk < 2) or (cjk == 0 and len(compact) < 3)): if compact and ((cjk == len(compact) and cjk < 2) or (cjk == 0 and len(compact) < 3)):
result["query_notice"] = ( result["query_notice"] = (
"关键词过短，仅进行精确人物、标签和标题匹配；" "关键词过短，仅进行精确人物、标签和标题匹配；搜索语音/OCR 至少输入 2 个汉字或 3 个拉丁字符。"
"搜索语音/OCR 至少输入 2 个汉字或 3 个拉丁字符。"
) )
return result return result
except ValueError as exc: except ValueError as exc:
@@ -2161,17 +2168,23 @@ def speech_rebuild_status(
if not isinstance(state, dict) or state.get("status") != "running": if not isinstance(state, dict) or state.get("status") != "running":
return state return state
with app.db.read() as conn: with app.db.read() as conn:
completed = int(conn.execute( completed = int(
"SELECT count(*) FROM videos WHERE available=1 AND audio_index_revision>=?", conn.execute(
(SPEECH_INDEX_REVISION,), "SELECT count(*) FROM videos WHERE available=1 AND audio_index_revision>=?",
).fetchone()[0]) (SPEECH_INDEX_REVISION,),
remaining = int(conn.execute( ).fetchone()[0]
"SELECT count(*) FROM videos WHERE available=1 AND audio_index_revision<?", )
(SPEECH_INDEX_REVISION,), remaining = int(
).fetchone()[0]) conn.execute(
active = int(conn.execute( "SELECT count(*) FROM videos WHERE available=1 AND audio_index_revision<?",
"SELECT count(*) FROM jobs WHERE kind='transcribe_audio' AND status IN ('queued','running')" (SPEECH_INDEX_REVISION,),
).fetchone()[0]) ).fetchone()[0]
)
active = int(
conn.execute(
"SELECT count(*) FROM jobs WHERE kind='transcribe_audio' AND status IN ('queued','running')"
).fetchone()[0]
)
state = {**state, "completed": completed, "remaining": remaining, "active_jobs": active} state = {**state, "completed": completed, "remaining": remaining, "active_jobs": active}
if remaining == 0 and active == 0: if remaining == 0 and active == 0:
state = {**state, "status": "completed", "finished_at": utcnow()} state = {**state, "status": "completed", "finished_at": utcnow()}
@@ -2201,9 +2214,9 @@ def video_transcript(
raise HTTPException(404, "视频不存在") raise HTTPException(404, "视频不存在")
total = int( total = int(
conn.execute( conn.execute(
"SELECT count(*) FROM text_entries WHERE video_id=? AND kind='audio' " "SELECT count(*) FROM text_entries WHERE video_id=? AND kind='audio' "
"AND coalesce(quality_score,1)>=0.55", "AND coalesce(quality_score,1)>=0.55",
(video_id,), (video_id,),
).fetchone()[0] ).fetchone()[0]
) )
rows = conn.execute( rows = conn.execute(
@@ -2658,9 +2671,7 @@ async def list_uploads(
source_id, source_id,
query, query,
) )
return await _background_api_with_fallback( return await _background_api_with_fallback(app.uploads.list, app.uploads.cached_list, limit)
app.uploads.list, app.uploads.cached_list, limit
)
@router.post("/v1/uploads", status_code=201) @router.post("/v1/uploads", status_code=201)
@@ -3032,9 +3043,7 @@ def restore_trash(
"ORDER BY updated_at DESC LIMIT 1", "ORDER BY updated_at DESC LIMIT 1",
(item["source_id"], item["original_key"], item["physical_original_path"]), (item["source_id"], item["original_key"], item["physical_original_path"]),
).fetchone() ).fetchone()
if item if item and item["storage_backend"] == "openlist_native" and item["physical_original_path"]
and item["storage_backend"] == "openlist_native"
and item["physical_original_path"]
else None else None
) )
key = app.storage.restore(trash_id) key = app.storage.restore(trash_id)
backend/imagefind/uploads.py
+14 -23
View File
@@ -86,14 +86,13 @@ class UploadService:
recovery_state = str(result.get("recovery_state") or "") recovery_state = str(result.get("recovery_state") or "")
if status in {"failed", "cancelled"} and recovery_state in {"", "receiving"}: if status in {"failed", "cancelled"} and recovery_state in {"", "receiving"}:
original = Path(str(result.get("temp_path") or "")).is_file() original = Path(str(result.get("temp_path") or "")).is_file()
encrypted = bool( encrypted = bool(result.get("external_local_path") and Path(str(result["external_local_path"])).is_file())
result.get("external_local_path")
and Path(str(result["external_local_path"])).is_file()
)
catalog = bool(result.get("failure_stage") == "catalog" and result.get("target_key")) catalog = bool(result.get("failure_stage") == "catalog" and result.get("target_key"))
recovery_state = "available" if original or encrypted else "catalog" if catalog else "missing" recovery_state = "available" if original or encrypted else "catalog" if catalog else "missing"
result["recovery_state"] = recovery_state result["recovery_state"] = recovery_state
result["recovery_mode"] = "original" if original else "encrypted" if encrypted else "catalog" if catalog else None result["recovery_mode"] = (
"original" if original else "encrypted" if encrypted else "catalog" if catalog else None
)
result["deduplicated"] = bool(result.get("deduplicated", False)) result["deduplicated"] = bool(result.get("deduplicated", False))
result.pop("temp_path", None) result.pop("temp_path", None)
result.pop("target_key", None) result.pop("target_key", None)
@@ -136,15 +135,15 @@ class UploadService:
local_done = phase not in {"receiving", "queued", "encrypting", "local_processing"} local_done = phase not in {"receiving", "queued", "encrypting", "local_processing"}
remote_required = result.get("storage_backend") == "openlist_native" remote_required = result.get("storage_backend") == "openlist_native"
remote_done = ( remote_done = (
(not remote_required) (not remote_required) or phase in {"cataloging", "ai_queued", "completed"} or status == "completed"
or phase in {"cataloging", "ai_queued", "completed"}
or status == "completed"
) )
catalog_done = phase in {"ai_queued", "completed"} or status == "completed" catalog_done = phase in {"ai_queued", "completed"} or status == "completed"
def step_state(done: bool, active: bool) -> str: def step_state(done: bool, active: bool) -> str:
if terminal_state: if terminal_state:
return terminal_state return terminal_state
return "completed" if done else "active" if active else "pending" return "completed" if done else "active" if active else "pending"
result["stages"] = [ result["stages"] = [
{ {
"key": "receiving", "key": "receiving",
@@ -427,6 +426,7 @@ class UploadService:
received = list(range(total_chunks)) if staged_path is not None else [] received = list(range(total_chunks)) if staged_path is not None else []
bytes_received = size_bytes if staged_path is not None else 0 bytes_received = size_bytes if staged_path is not None else 0
try: try:
def insert_upload(conn): def insert_upload(conn):
conn.execute( conn.execute(
"INSERT INTO uploads(id,source_id,relative_path,filename,title,collection_id,collection_parent_id," "INSERT INTO uploads(id,source_id,relative_path,filename,title,collection_id,collection_parent_id,"
@@ -632,6 +632,7 @@ class UploadService:
missing = sorted(set(range(upload["total_chunks"])) - set(upload["received_chunks"])) missing = sorted(set(range(upload["total_chunks"])) - set(upload["received_chunks"]))
if missing: if missing:
return {"completed": False, "missing_chunks": missing} return {"completed": False, "missing_chunks": missing}
def queue_transfer(conn): def queue_transfer(conn):
job_id = self.jobs.enqueue( job_id = self.jobs.enqueue(
"transfer_upload", "transfer_upload",
@@ -685,9 +686,7 @@ class UploadService:
def _refresh_recovery_state(self, upload_id: str) -> str: def _refresh_recovery_state(self, upload_id: str) -> str:
upload = self._get(upload_id) upload = self._get(upload_id)
original = Path(upload["temp_path"]).is_file() original = Path(upload["temp_path"]).is_file()
encrypted = bool( encrypted = bool(upload.get("external_local_path") and Path(str(upload["external_local_path"])).is_file())
upload.get("external_local_path") and Path(str(upload["external_local_path"])).is_file()
)
catalog = bool(upload.get("failure_stage") == "catalog" and upload.get("target_key")) catalog = bool(upload.get("failure_stage") == "catalog" and upload.get("target_key"))
state = "available" if original or encrypted else "catalog" if catalog else "missing" state = "available" if original or encrypted else "catalog" if catalog else "missing"
mode = "original" if original else "encrypted" if encrypted else "catalog" if catalog else None mode = "original" if original else "encrypted" if encrypted else "catalog" if catalog else None
@@ -706,9 +705,7 @@ class UploadService:
if upload["status"] not in {"failed", "cancelled", "completed"}: if upload["status"] not in {"failed", "cancelled", "completed"}:
raise ValueError("任务仍在执行，请先取消") raise ValueError("任务仍在执行，请先取消")
if upload.get("storage_backend") == "openlist_native" and self.openlist_native is not None: if upload.get("storage_backend") == "openlist_native" and self.openlist_native is not None:
self.openlist_native.cleanup( self.openlist_native.cleanup(upload["source_id"], upload_id, upload.get("external_staged_path"))
upload["source_id"], upload_id, upload.get("external_staged_path")
)
Path(upload["temp_path"]).unlink(missing_ok=True) Path(upload["temp_path"]).unlink(missing_ok=True)
with self.db.transaction() as conn: with self.db.transaction() as conn:
conn.execute("DELETE FROM ingest_guards WHERE upload_id=?", (upload_id,)) conn.execute("DELETE FROM ingest_guards WHERE upload_id=?", (upload_id,))
@@ -736,9 +733,7 @@ class UploadService:
) )
else: else:
staged_local = ( staged_local = (
upload.get("external_local_path") upload.get("external_local_path") if upload.get("storage_backend") == "openlist_native" else None
if upload.get("storage_backend") == "openlist_native"
else None
) )
if not Path(upload["temp_path"]).is_file() and not (staged_local and Path(staged_local).is_file()): if not Path(upload["temp_path"]).is_file() and not (staged_local and Path(staged_local).is_file()):
raise ValueError("上传暂存文件不存在，无法重试") raise ValueError("上传暂存文件不存在，无法重试")
@@ -1058,9 +1053,7 @@ class UploadService:
if not upload.get("external_source_path"): if not upload.get("external_source_path"):
path = Path(upload["temp_path"]) path = Path(upload["temp_path"])
paths = self.openlist_native.prepare( paths = self.openlist_native.prepare(upload["source_id"], upload_id, str(target_key), path)
upload["source_id"], upload_id, str(target_key), path
)
self.db.write_with_retry( self.db.write_with_retry(
lambda conn: conn.execute( lambda conn: conn.execute(
"UPDATE uploads SET status='transferring',phase='external_copying'," "UPDATE uploads SET status='transferring',phase='external_copying',"
@@ -1469,9 +1462,7 @@ class UploadService:
temp_path = Path(upload["temp_path"]) temp_path = Path(upload["temp_path"])
external_local = upload.get("external_local_path") external_local = upload.get("external_local_path")
recoverable_native = bool( recoverable_native = bool(
upload.get("storage_backend") == "openlist_native" upload.get("storage_backend") == "openlist_native" and external_local and Path(external_local).is_file()
and external_local
and Path(external_local).is_file()
) )
if not committed and not temp_path.is_file() and not recoverable_native: if not committed and not temp_path.is_file() and not recoverable_native:
expires = datetime.now(UTC) + timedelta(hours=self.settings.upload_failed_hours) expires = datetime.now(UTC) + timedelta(hours=self.settings.upload_failed_hours)