支持导入 PixivBatchDownloader 导出的抓取结果 JSON,该格式默认不使用补全

This commit is contained in:
inliver233 committed 2026-02-24 18:14:28 +08:00
1 parent 9e437d86b2
commit ef66583b9c
6 files changed
+714 -76

No files matched your search

+118 -15
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
import json
import os
from dataclasses import asdict, dataclass
from typing import Any, Literal
from typing import Any, Literal, cast
from uuid import uuid4
import sqlalchemy as sa
@@ -40,6 +40,13 @@ class ImportRollbackRequest(BaseModel):
mode: Literal["disable", "delete"]
@dataclass(frozen=True, slots=True)
class ImportUpload:
body: ImportCreateRequest
payload_bytes: bytes
input_format: Literal["text", "pixiv_batch_downloader_json"]
@dataclass(frozen=True, slots=True)
class ImportErrorItem:
line: int
@@ -49,7 +56,8 @@ class ImportErrorItem:
def _max_import_text_bytes() -> int:
default = 50 * 1024 * 1024
default = 200 * 1024 * 1024
cap = 600 * 1024 * 1024
raw = (os.environ.get("IMPORT_MAX_BYTES") or "").strip()
if not raw:
return default
@@ -57,7 +65,7 @@ def _max_import_text_bytes() -> int:
value = int(raw)
except Exception:
return default
return max(1024, min(int(value), 50 * 1024 * 1024))
return max(1024, min(int(value), int(cap)))
def _import_inline_max_accepted() -> int:
@@ -166,7 +174,86 @@ def _validate_import_create(data: dict[str, Any]) -> ImportCreateRequest:
raise ApiError(code=ErrorCode.BAD_REQUEST, message="导入请求体无效", status_code=400) from exc
async def _load_import_request(request: Request) -> ImportCreateRequest:
def _parse_pixiv_batch_downloader_json(
raw: bytes, *, preview_limit: int = 20
) -> tuple[int, int, int, int, list[ImportErrorItem], list[dict[str, Any]]]:
try:
text = raw.decode("utf-8-sig", errors="replace")
except Exception:
text = raw.decode("utf-8", errors="replace")
try:
parsed = json.loads(text)
except json.JSONDecodeError as exc:
raise ApiError(code=ErrorCode.BAD_REQUEST, message="JSON 文件解析失败", status_code=400) from exc
items: list[Any] | None = None
if isinstance(parsed, list):
items = parsed
elif isinstance(parsed, dict):
if isinstance(parsed.get("result"), list):
items = cast(list[Any], parsed.get("result"))
elif isinstance(parsed.get("data"), list):
items = cast(list[Any], parsed.get("data"))
if items is None:
raise ApiError(code=ErrorCode.BAD_REQUEST, message="不支持的 JSON 格式(需要数组)", status_code=400)
total = 0
accepted = 0
deduped = 0
error_total = 0
errors: list[ImportErrorItem] = []
seen: set[tuple[int, int]] = set()
preview: list[dict[str, Any]] = []
for idx, raw_item in enumerate(items, start=1):
if not isinstance(raw_item, dict):
continue
url = raw_item.get("original")
if not isinstance(url, str):
continue
url = url.strip()
if not url:
continue
total += 1
try:
parsed_url = parse_pixiv_original_url(url)
except Exception as exc:
error_total += 1
if len(errors) < 200:
errors.append(
ImportErrorItem(
line=idx,
url=url,
code=ErrorCode.UNSUPPORTED_URL.value,
message=str(exc) or ErrorCode.UNSUPPORTED_URL.value,
)
)
continue
key = (parsed_url.illust_id, parsed_url.page_index)
if key in seen:
deduped += 1
continue
seen.add(key)
accepted += 1
if len(preview) < int(preview_limit):
preview.append(
{
"illust_id": parsed_url.illust_id,
"page_index": parsed_url.page_index,
"ext": parsed_url.ext,
"url": url,
}
)
return total, accepted, deduped, error_total, errors, preview
async def _load_import_request(request: Request) -> ImportUpload:
content_type = (request.headers.get("content-type") or "").lower()
max_bytes = _max_import_text_bytes()
@@ -175,9 +262,10 @@ async def _load_import_request(request: Request) -> ImportCreateRequest:
if not isinstance(data, dict):
raise ApiError(code=ErrorCode.BAD_REQUEST, message="Invalid JSON body", status_code=400)
body = _validate_import_create(data)
if len(body.text.encode("utf-8", errors="ignore")) > max_bytes:
payload_bytes = body.text.encode("utf-8", errors="ignore")
if len(payload_bytes) > max_bytes:
raise ApiError(code=ErrorCode.PAYLOAD_TOO_LARGE, message="Payload too large", status_code=413)
return body
return ImportUpload(body=body, payload_bytes=payload_bytes, input_format="text")
if content_type.startswith("multipart/form-data"):
form = await request.form()
@@ -186,22 +274,25 @@ async def _load_import_request(request: Request) -> ImportCreateRequest:
raise ApiError(code=ErrorCode.BAD_REQUEST, message="Missing file", status_code=400)
filename = (file_obj.filename or "").strip().lower()
if filename and not filename.endswith(".txt"):
input_format: Literal["text", "pixiv_batch_downloader_json"] = "text"
if filename.endswith(".json"):
input_format = "pixiv_batch_downloader_json"
if filename and not (filename.endswith(".txt") or filename.endswith(".json")):
raise ApiError(code=ErrorCode.INVALID_UPLOAD_TYPE, message="Unsupported upload type", status_code=400)
raw = await file_obj.read()
if len(raw) > max_bytes:
raise ApiError(code=ErrorCode.PAYLOAD_TOO_LARGE, message="Payload too large", status_code=413)
text = raw.decode("utf-8", errors="replace")
return _validate_import_create(
body = _validate_import_create(
{
"text": text,
"text": "pixiv_batch_downloader_json" if input_format == "pixiv_batch_downloader_json" else raw.decode("utf-8", errors="replace"),
"dry_run": _parse_bool(form.get("dry_run"), default=False),
"hydrate_on_import": _parse_bool(form.get("hydrate_on_import"), default=False),
"source": str(form.get("source") or "manual"),
}
)
return ImportUpload(body=body, payload_bytes=raw, input_format=input_format)
raise ApiError(code=ErrorCode.INVALID_UPLOAD_TYPE, message="Unsupported content type", status_code=400)
@@ -211,10 +302,16 @@ async def create_import(
request: Request,
_claims: dict[str, Any] = Depends(get_admin_claims),
) -> dict[str, Any]:
body = await _load_import_request(request)
upload = await _load_import_request(request)
body = upload.body
rid = get_or_create_request_id(request)
total, accepted, deduped, error_total, errors, preview = _parse_import_text(body.text, preview_limit=20)
if upload.input_format == "pixiv_batch_downloader_json":
total, accepted, deduped, error_total, errors, preview = _parse_pixiv_batch_downloader_json(
upload.payload_bytes, preview_limit=20
)
else:
total, accepted, deduped, error_total, errors, preview = _parse_import_text(body.text, preview_limit=20)
if body.dry_run:
return {
@@ -231,14 +328,19 @@ async def create_import(
engine = request.app.state.engine
Session = create_sessionmaker(engine)
hydrate_on_import = bool(body.hydrate_on_import)
if upload.input_format == "pixiv_batch_downloader_json":
hydrate_on_import = False
settings = request.app.state.settings
db_dir = get_sqlite_db_dir(settings.database_url)
payload_dir = db_dir / "imports_payloads"
payload_dir.mkdir(parents=True, exist_ok=True)
payload_path = payload_dir / f"import_payload_{uuid4().hex}.txt"
ext = ".json" if upload.input_format == "pixiv_batch_downloader_json" else ".txt"
payload_path = payload_dir / f"import_payload_{uuid4().hex}{ext}"
try:
payload_path.write_text(body.text, encoding="utf-8")
payload_path.write_bytes(upload.payload_bytes)
except OSError as exc:
raise ApiError(code=ErrorCode.INTERNAL_ERROR, message="写入导入内容失败", status_code=500) from exc
@@ -269,7 +371,8 @@ async def create_import(
{
"import_id": int(imp.id),
"file_ref": file_ref,
"hydrate_on_import": bool(body.hydrate_on_import),
"hydrate_on_import": bool(hydrate_on_import),
"input_format": upload.input_format,
},
ensure_ascii=False,
separators=(",", ":"),
+349 -50
View File
@@ -4,6 +4,7 @@ import json
import random
from collections.abc import Iterable
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
@@ -14,9 +15,11 @@ from sqlalchemy.ext.asyncio import AsyncEngine
from app.core.config import load_settings
from app.core.data_files import get_sqlite_db_dir, resolve_file_ref
from app.core.pixiv_urls import parse_pixiv_original_url
from app.db.models.image_tags import ImageTag
from app.db.models.images import Image
from app.db.models.imports import Import
from app.db.models.jobs import JobRow
from app.db.models.tags import Tag
from app.db.session import create_sessionmaker, with_sqlite_busy_retry
from app.jobs.errors import JobPermanentError
@@ -48,6 +51,71 @@ def _as_bool(value: Any, *, default: bool = False) -> bool:
return default
def _as_int(value: Any) -> int | None:
if value is None:
return None
if isinstance(value, bool):
return None
try:
return int(value)
except Exception:
return None
def _as_str(value: Any) -> str | None:
if value is None:
return None
s = str(value).strip()
return s if s else None
def _parse_pbd_ai_type(value: Any) -> int | None:
raw = _as_int(value)
if raw is None:
return None
# PixivBatchDownloader: 0 unknown, 1 non-ai, 2 ai
if raw == 1:
return 0
if raw == 2:
return 1
return None
def _parse_pbd_illust_type(value: Any) -> int | None:
raw = _as_int(value)
if raw in {0, 1, 2}:
return int(raw)
return None
def _parse_pbd_created_at(value: Any) -> str | None:
s = _as_str(value)
if not s:
return None
try:
dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
except Exception:
if len(s) <= 64:
return s
return None
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
dt_utc = dt.astimezone(timezone.utc).replace(microsecond=0)
return dt_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
def _derive_orientation(width: int | None, height: int | None) -> tuple[float | None, int | None]:
if width is None or height is None or width <= 0 or height <= 0:
return None, None
if width > height:
orientation = 2
elif height > width:
orientation = 1
else:
orientation = 3
return float(width) / float(height), orientation
def _resolve_payload_file(payload: dict[str, Any]) -> Path | None:
if "file_ref" not in payload:
return None
@@ -115,8 +183,17 @@ def build_import_images_handler(engine: AsyncEngine):
if import_id <= 0:
raise JobPermanentError("payload.import_id is required")
input_format_raw = str(payload.get("input_format") or "text").strip().lower()
input_format = "pixiv_batch_downloader_json" if input_format_raw in {"pixiv_batch_downloader_json", "pbd_json", "pbd"} else "text"
hydrate_on_import = _as_bool(payload.get("hydrate_on_import"), default=False)
if input_format == "pixiv_batch_downloader_json":
# PixivBatchDownloader export already contains most metadata;
# keep this import token-free by default.
hydrate_on_import = False
file_path = _resolve_payload_file(payload)
if input_format == "pixiv_batch_downloader_json" and file_path is None:
raise JobPermanentError("payload.file_ref is required for pixiv_batch_downloader_json")
async def _ensure_import_exists() -> None:
async with Session() as session:
@@ -137,6 +214,7 @@ def build_import_images_handler(engine: AsyncEngine):
chunk_rows: list[dict[str, Any]] = []
chunk_keys: list[tuple[int, int]] = []
chunk_tags: dict[tuple[int, int], list[str]] = {}
async def _persist_chunk(
rows: list[dict[str, Any]],
@@ -146,6 +224,7 @@ def build_import_images_handler(engine: AsyncEngine):
accepted_v: int,
success_v: int,
failed_v: int,
tags_by_key: dict[tuple[int, int], list[str]] | None = None,
) -> None:
if not rows:
return
@@ -165,6 +244,47 @@ def build_import_images_handler(engine: AsyncEngine):
else_=Image.proxy_path,
),
"created_import_id": stmt.excluded.created_import_id,
"width": sa.case((stmt.excluded.width.is_not(None), stmt.excluded.width), else_=Image.width),
"height": sa.case((stmt.excluded.height.is_not(None), stmt.excluded.height), else_=Image.height),
"aspect_ratio": sa.case(
(stmt.excluded.aspect_ratio.is_not(None), stmt.excluded.aspect_ratio),
else_=Image.aspect_ratio,
),
"orientation": sa.case(
(stmt.excluded.orientation.is_not(None), stmt.excluded.orientation),
else_=Image.orientation,
),
"x_restrict": sa.case(
(stmt.excluded.x_restrict.is_not(None), stmt.excluded.x_restrict),
else_=Image.x_restrict,
),
"ai_type": sa.case((stmt.excluded.ai_type.is_not(None), stmt.excluded.ai_type), else_=Image.ai_type),
"illust_type": sa.case(
(stmt.excluded.illust_type.is_not(None), stmt.excluded.illust_type),
else_=Image.illust_type,
),
"user_id": sa.case((stmt.excluded.user_id.is_not(None), stmt.excluded.user_id), else_=Image.user_id),
"user_name": sa.case(
(stmt.excluded.user_name.is_not(None), stmt.excluded.user_name),
else_=Image.user_name,
),
"title": sa.case((stmt.excluded.title.is_not(None), stmt.excluded.title), else_=Image.title),
"created_at_pixiv": sa.case(
(stmt.excluded.created_at_pixiv.is_not(None), stmt.excluded.created_at_pixiv),
else_=Image.created_at_pixiv,
),
"bookmark_count": sa.case(
(stmt.excluded.bookmark_count.is_not(None), stmt.excluded.bookmark_count),
else_=Image.bookmark_count,
),
"view_count": sa.case(
(stmt.excluded.view_count.is_not(None), stmt.excluded.view_count),
else_=Image.view_count,
),
"comment_count": sa.case(
(stmt.excluded.comment_count.is_not(None), stmt.excluded.comment_count),
else_=Image.comment_count,
),
"updated_at": now_expr,
},
)
@@ -180,6 +300,63 @@ def build_import_images_handler(engine: AsyncEngine):
.values(proxy_path=sa.text("'/i/' || id || '.' || ext"))
)
if tags_by_key and keys:
names: list[str] = []
seen_names: set[str] = set()
for lst in tags_by_key.values():
for raw_name in lst[:64]:
name = str(raw_name or "").strip()
if not name or name in seen_names:
continue
seen_names.add(name)
names.append(name)
if names:
tag_stmt = sqlite_insert(Tag).values([{"name": n, "translated_name": None} for n in names])
tag_stmt = tag_stmt.on_conflict_do_nothing(index_elements=["name"])
await session.execute(tag_stmt)
tag_rows = (
(await session.execute(sa.select(Tag.id, Tag.name).where(Tag.name.in_(names))))
.all()
)
tag_id_by_name = {str(name): int(tag_id) for (tag_id, name) in tag_rows}
img_rows = (
(
await session.execute(
sa.select(Image.id, Image.illust_id, Image.page_index).where(
sa.tuple_(Image.illust_id, Image.page_index).in_(keys)
)
)
)
.all()
)
image_id_by_key = {(int(illust_id), int(page_index)): int(img_id) for (img_id, illust_id, page_index) in img_rows}
image_tag_rows: list[dict[str, Any]] = []
for key, lst in tags_by_key.items():
image_id = image_id_by_key.get(key)
if image_id is None:
continue
seen_for_image: set[str] = set()
for raw_name in lst[:64]:
name = str(raw_name or "").strip()
if not name or name in seen_for_image:
continue
seen_for_image.add(name)
tag_id = tag_id_by_name.get(name)
if tag_id is None:
continue
image_tag_rows.append({"image_id": int(image_id), "tag_id": int(tag_id)})
if image_tag_rows:
for offset in range(0, len(image_tag_rows), 5000):
sub = image_tag_rows[offset : offset + 5000]
it_stmt = sqlite_insert(ImageTag).values(sub)
it_stmt = it_stmt.on_conflict_do_nothing(index_elements=["image_id", "tag_id"])
await session.execute(it_stmt)
await session.execute(
sa.update(Import)
.where(Import.id == int(import_id))
@@ -194,64 +371,184 @@ def build_import_images_handler(engine: AsyncEngine):
await with_sqlite_busy_retry(_op)
for line_no, raw in _iter_lines(payload, file_path=file_path):
url = raw.strip()
if not url:
continue
total += 1
if input_format == "pixiv_batch_downloader_json":
try:
parsed = parse_pixiv_original_url(url)
with file_path.open("r", encoding="utf-8-sig", errors="replace") as f:
data = json.load(f)
except Exception as exc:
error_total += 1
if len(errors) < _MAX_ERRORS:
errors.append(
ImportLineError(
line=int(line_no),
url=url,
code="unsupported_url",
message=str(exc) or "unsupported_url",
raise JobPermanentError("payload.file_ref is not valid JSON") from exc
items: list[Any] | None = None
if isinstance(data, list):
items = data
elif isinstance(data, dict):
if isinstance(data.get("result"), list):
items = data.get("result")
elif isinstance(data.get("data"), list):
items = data.get("data")
if items is None:
raise JobPermanentError("payload.file_ref has unsupported JSON shape")
for idx, raw_item in enumerate(items, start=1):
if not isinstance(raw_item, dict):
continue
url = _as_str(raw_item.get("original"))
if not url:
continue
total += 1
try:
parsed = parse_pixiv_original_url(url)
except Exception as exc:
error_total += 1
if len(errors) < _MAX_ERRORS:
errors.append(
ImportLineError(
line=int(idx),
url=url,
code="unsupported_url",
message=str(exc) or "unsupported_url",
)
)
)
continue
continue
key = (int(parsed.illust_id), int(parsed.page_index))
if key in seen:
deduped += 1
continue
seen.add(key)
key = (int(parsed.illust_id), int(parsed.page_index))
if key in seen:
deduped += 1
continue
seen.add(key)
accepted += 1
if hydrate_on_import:
illust_ids.add(int(parsed.illust_id))
accepted += 1
chunk_keys.append(key)
chunk_rows.append(
{
"illust_id": int(parsed.illust_id),
"page_index": int(parsed.page_index),
"ext": str(parsed.ext),
"original_url": url,
"proxy_path": "",
"random_key": float(random.random()),
"created_import_id": int(import_id),
}
)
width = _as_int(raw_item.get("fullWidth"))
height = _as_int(raw_item.get("fullHeight"))
if width is not None and width <= 0:
width = None
if height is not None and height <= 0:
height = None
if len(chunk_rows) < _CHUNK_SIZE:
continue
aspect_ratio, orientation = _derive_orientation(width, height)
x_restrict = _as_int(raw_item.get("xRestrict"))
if x_restrict not in {0, 1, 2}:
x_restrict = None
success_after = int(success + len(chunk_rows))
await _persist_chunk(
chunk_rows,
chunk_keys,
total_v=int(total),
accepted_v=int(accepted),
success_v=int(success_after),
failed_v=int(error_total),
)
success = success_after
chunk_rows.clear()
chunk_keys.clear()
chunk_keys.append(key)
chunk_rows.append(
{
"illust_id": int(parsed.illust_id),
"page_index": int(parsed.page_index),
"ext": str(parsed.ext),
"original_url": url,
"proxy_path": "",
"random_key": float(random.random()),
"created_import_id": int(import_id),
"width": width,
"height": height,
"aspect_ratio": aspect_ratio,
"orientation": orientation,
"x_restrict": x_restrict,
"ai_type": _parse_pbd_ai_type(raw_item.get("aiType")),
"illust_type": _parse_pbd_illust_type(raw_item.get("type")),
"user_id": _as_int(raw_item.get("userId")),
"user_name": _as_str(raw_item.get("user")),
"title": _as_str(raw_item.get("title")),
"created_at_pixiv": _parse_pbd_created_at(raw_item.get("date")),
"bookmark_count": _as_int(raw_item.get("bmk")),
"view_count": _as_int(raw_item.get("viewCount")),
"comment_count": _as_int(raw_item.get("commentCount")),
}
)
raw_tags = raw_item.get("tags")
if isinstance(raw_tags, list) and raw_tags:
tags: list[str] = []
seen_tags: set[str] = set()
for v in raw_tags[:128]:
name = str(v or "").strip()
if not name or name in seen_tags:
continue
seen_tags.add(name)
tags.append(name)
if len(tags) >= 64:
break
if tags:
chunk_tags[key] = tags
if len(chunk_rows) < _CHUNK_SIZE:
continue
success_after = int(success + len(chunk_rows))
await _persist_chunk(
chunk_rows,
chunk_keys,
total_v=int(total),
accepted_v=int(accepted),
success_v=int(success_after),
failed_v=int(error_total),
tags_by_key=dict(chunk_tags),
)
success = success_after
chunk_rows.clear()
chunk_keys.clear()
chunk_tags.clear()
else:
for line_no, raw in _iter_lines(payload, file_path=file_path):
url = raw.strip()
if not url:
continue
total += 1
try:
parsed = parse_pixiv_original_url(url)
except Exception as exc:
error_total += 1
if len(errors) < _MAX_ERRORS:
errors.append(
ImportLineError(
line=int(line_no),
url=url,
code="unsupported_url",
message=str(exc) or "unsupported_url",
)
)
continue
key = (int(parsed.illust_id), int(parsed.page_index))
if key in seen:
deduped += 1
continue
seen.add(key)
accepted += 1
if hydrate_on_import:
illust_ids.add(int(parsed.illust_id))
chunk_keys.append(key)
chunk_rows.append(
{
"illust_id": int(parsed.illust_id),
"page_index": int(parsed.page_index),
"ext": str(parsed.ext),
"original_url": url,
"proxy_path": "",
"random_key": float(random.random()),
"created_import_id": int(import_id),
}
)
if len(chunk_rows) < _CHUNK_SIZE:
continue
success_after = int(success + len(chunk_rows))
await _persist_chunk(
chunk_rows,
chunk_keys,
total_v=int(total),
accepted_v=int(accepted),
success_v=int(success_after),
failed_v=int(error_total),
)
success = success_after
chunk_rows.clear()
chunk_keys.clear()
if chunk_rows:
success_after = int(success + len(chunk_rows))
@@ -262,10 +559,12 @@ def build_import_images_handler(engine: AsyncEngine):
accepted_v=int(accepted),
success_v=int(success_after),
failed_v=int(error_total),
tags_by_key=dict(chunk_tags) if chunk_tags else None,
)
success = success_after
chunk_rows.clear()
chunk_keys.clear()
chunk_tags.clear()
async def _persist_detail() -> None:
async with Session() as session:
+95 -1
View File
@@ -142,7 +142,101 @@ def test_admin_imports_multipart_happy_path(tmp_path: Path, monkeypatch) -> None
assert row is not None
return (int(row[0]), int(row[1]), int(row[2]), int(row[3]))
assert asyncio.run(_fetch_import_counts()) == (4, 2, 2, 1)
assert asyncio.run(_fetch_import_counts()) == (4, 2, 2, 1)
def test_admin_imports_multipart_pixiv_batch_downloader_json_imports_metadata(tmp_path: Path, monkeypatch) -> None:
db_path = tmp_path / "admin_imports_pbd_json.db"
db_url = "sqlite+aiosqlite:///" + db_path.as_posix()
monkeypatch.setenv("APP_ENV", "dev")
monkeypatch.setenv("DATABASE_URL", db_url)
monkeypatch.setenv("SECRET_KEY", "secret_test")
monkeypatch.setenv("ADMIN_USERNAME", "admin")
app = create_app()
async def _migrate() -> None:
async with app.state.engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
asyncio.run(_migrate())
token = create_jwt(secret_key="secret_test", subject="admin", ttl_s=3600)
with TestClient(app) as client:
payload = (
b"["
+ b"{"
+ b"\"id\":\"12345678_p0\","
+ b"\"index\":0,"
+ b"\"original\":\"https://i.pximg.net/img-original/img/2023/01/01/00/00/00/12345678_p0.jpg\","
+ b"\"fullWidth\":1000,\"fullHeight\":1500,"
+ b"\"xRestrict\":0,\"aiType\":1,\"type\":0,"
+ b"\"userId\":\"42\",\"user\":\"alice\",\"title\":\"hello\","
+ b"\"date\":\"2023-01-01T00:00:00+09:00\","
+ b"\"bmk\":12,\"viewCount\":34,\"commentCount\":5,"
+ b"\"tags\":[\"loli\",\"cute\"]"
+ b"}"
+ b"]"
)
resp = client.post(
"/admin/api/imports",
headers={"Authorization": f"Bearer {token}", "X-Request-Id": "req_test"},
data={"dry_run": "false", "hydrate_on_import": "true", "source": "manual"},
files={"file": ("result.json", payload, "application/json")},
)
assert resp.status_code == 200
body = resp.json()
assert body["ok"] is True
assert body["accepted"] == 1
assert body["deduped"] == 0
assert body["errors"] == []
assert body["import_id"].isdigit()
assert body["job_id"].isdigit()
executed_inline = bool(body.get("executed_inline"))
import_id = int(body["import_id"])
if not executed_inline:
async def _run_worker() -> None:
dispatcher = build_default_dispatcher(app.state.engine)
ran = await poll_and_execute_jobs(app.state.engine, dispatcher, worker_id="test-worker", max_jobs=10)
assert ran >= 1
asyncio.run(_run_worker())
async def _fetch_image_and_tag_counts() -> tuple[int, int, int]:
async with app.state.engine.connect() as conn:
img = await conn.exec_driver_sql(
"SELECT width,height,orientation,ai_type,x_restrict,user_id,bookmark_count FROM images WHERE illust_id=12345678 AND page_index=0"
)
row = img.fetchone()
assert row is not None
tag_count = int((await conn.exec_driver_sql("SELECT COUNT(*) FROM tags")).scalar_one())
link_count = int((await conn.exec_driver_sql("SELECT COUNT(*) FROM image_tags")).scalar_one())
return (int(row[0]), int(row[1]), tag_count, link_count)
assert asyncio.run(_fetch_image_and_tag_counts()) == (1000, 1500, 2, 2)
async def _hydrate_job_count() -> int:
async with app.state.engine.connect() as conn:
return int((await conn.exec_driver_sql("SELECT COUNT(*) FROM jobs WHERE type='hydrate_metadata'")).scalar_one())
assert asyncio.run(_hydrate_job_count()) == 0
async def _fetch_import_counts() -> tuple[int, int, int, int]:
async with app.state.engine.connect() as conn:
result = await conn.exec_driver_sql(
"SELECT total, accepted, success, failed FROM imports WHERE id = ?",
(import_id,),
)
row = result.fetchone()
assert row is not None
return (int(row[0]), int(row[1]), int(row[2]), int(row[3]))
assert asyncio.run(_fetch_import_counts()) == (1, 1, 1, 0)
def test_admin_imports_dry_run_preview_does_not_write_db(tmp_path: Path, monkeypatch) -> None:
@@ -8,9 +8,11 @@ import sqlalchemy as sa
from app.db.engine import create_engine
from app.db.models.base import Base
from app.db.models.image_tags import ImageTag
from app.db.models.images import Image
from app.db.models.imports import Import
from app.db.models.jobs import JobRow
from app.db.models.tags import Tag
from app.db.session import create_sessionmaker
from app.jobs.claim import claim_next_job
from app.jobs.dispatch import JobDispatcher
@@ -274,6 +276,127 @@ def test_job_handler_import_images_supports_file_ref(tmp_path: Path, monkeypatch
asyncio.run(_run())
def test_job_handler_import_images_supports_pixiv_batch_downloader_json(tmp_path: Path, monkeypatch) -> None:
db_path = tmp_path / "handler_import_images_pbd_json.db"
db_url = _sqlite_url(db_path)
engine = create_engine(db_url)
monkeypatch.setenv("APP_ENV", "dev")
monkeypatch.setenv("DATABASE_URL", db_url)
payload_dir = tmp_path / "imports_payloads"
payload_dir.mkdir(parents=True, exist_ok=True)
payload_path = payload_dir / "pbd.json"
payload_path.write_text(
json.dumps(
[
{
"id": "12345678_p0",
"index": 0,
"original": "https://i.pximg.net/img-original/img/2023/01/01/00/00/00/12345678_p0.jpg",
"fullWidth": 1000,
"fullHeight": 1500,
"xRestrict": 1,
"aiType": 2,
"type": 0,
"userId": "42",
"user": "alice",
"title": "hello",
"date": "2023-01-01T00:00:00+09:00",
"bmk": 12,
"viewCount": 34,
"commentCount": 5,
"tags": ["loli", "cute", "loli"],
}
],
ensure_ascii=False,
separators=(",", ":"),
),
encoding="utf-8",
)
async def _run() -> None:
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
Session = create_sessionmaker(engine)
async with Session() as session:
imp = Import(created_by="admin", source="pixiv_batch_downloader")
session.add(imp)
await session.commit()
await session.refresh(imp)
payload = {
"import_id": int(imp.id),
"hydrate_on_import": True, # should be ignored for pbd json
"input_format": "pixiv_batch_downloader_json",
"file_ref": "imports_payloads/pbd.json",
}
session.add(
JobRow(
type="import_images",
status="pending",
payload_json=json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
ref_type="import",
ref_id=str(int(imp.id)),
)
)
await session.commit()
dispatcher = JobDispatcher()
dispatcher.register("import_images", build_import_images_handler(engine))
claimed = await claim_next_job(engine, worker_id="w1")
assert claimed is not None
transition = await execute_claimed_job(engine, dispatcher, job_row=claimed, worker_id="w1")
assert transition is not None
assert transition.status.value == "completed"
async with Session() as session:
imgs = ((await session.execute(sa.select(Image))).scalars().all())
assert len(imgs) == 1
img = imgs[0]
assert img.width == 1000
assert img.height == 1500
assert img.orientation == 1
assert img.aspect_ratio is not None
assert abs(float(img.aspect_ratio) - (1000.0 / 1500.0)) < 1e-6
assert img.x_restrict == 1
assert img.ai_type == 1
assert img.illust_type == 0
assert img.user_id == 42
assert img.user_name == "alice"
assert img.title == "hello"
assert img.created_at_pixiv == "2022-12-31T15:00:00Z"
assert img.bookmark_count == 12
assert img.view_count == 34
assert img.comment_count == 5
tags = ((await session.execute(sa.select(Tag).order_by(Tag.name.asc()))).scalars().all())
assert [t.name for t in tags] == ["cute", "loli"]
links = ((await session.execute(sa.select(ImageTag))).scalars().all())
assert len(links) == 2
hydrate_jobs = (
(
await session.execute(
sa.select(JobRow).where(JobRow.type == "hydrate_metadata")
)
)
.scalars()
.all()
)
assert hydrate_jobs == []
assert payload_path.exists() is False
await engine.dispose()
asyncio.run(_run())
def test_job_handler_import_images_missing_import_id_moves_to_dlq(tmp_path: Path) -> None:
db_path = tmp_path / "handler_import_images_missing_import_id.db"
engine = create_engine(_sqlite_url(db_path))
+25 -6
View File
@@ -74,6 +74,7 @@ export function ImportPage() {
const [errorMessage, setErrorMessage] = useState<string | null>(null);
const [requestId, setRequestId] = useState<string | null>(null);
const [result, setResult] = useState<ImportCreateResponse | null>(null);
const fileIsJson = Boolean(file && String(file.name || "").toLowerCase().endsWith(".json"));
const mutation = useMutation({
mutationFn: (values: ImportFormValues) => {
@@ -122,12 +123,14 @@ export function ImportPage() {
<Card title="支持的链接格式(重要)">
<Space direction="vertical" size={4}>
<Typography.Text>仅支持 Pixiv 原图(pximg)链接,每行一个:</Typography.Text>
<Typography.Text>支持两种导入方式:</Typography.Text>
<Typography.Text>- 纯链接:仅支持 Pixiv 原图(pximg)链接,每行一个:</Typography.Text>
<Typography.Text code copyable>
https://i.pximg.net/img-original/img/2024/01/01/00/00/00/12345678_p0.png
</Typography.Text>
<Typography.Text type="secondary">多P作品需要分别提供 p0/p1/... 的链接。</Typography.Text>
<Typography.Text type="secondary">不支持作品页链接(例如 www.pixiv.net/artworks/12345678)。</Typography.Text>
<Typography.Text>- PixivBatchDownloader:支持直接上传其导出的 <Typography.Text code>.json</Typography.Text>(自动提取原图链接,并尽可能写入元数据/标签,通常不需要 refresh token)。</Typography.Text>
</Space>
</Card>
@@ -138,16 +141,28 @@ export function ImportPage() {
initialValues={{ dry_run: false, hydrate_on_import: true, text: "" }}
onFinish={(values) => mutation.mutate(values)}
>
<Form.Item label="上传文本文件(可选,.txt)">
<Form.Item label="上传文件(可选,.txt / PixivBatchDownloader .json)">
<input
data-testid="import-file-input"
type="file"
accept=".txt,text/plain"
onChange={(e) => setFile(e.target.files && e.target.files[0] ? e.target.files[0] : null)}
accept=".txt,.json,text/plain,application/json"
onChange={(e) => {
const picked = e.target.files && e.target.files[0] ? e.target.files[0] : null;
setFile(picked);
const isJson = Boolean(picked && String(picked.name || "").toLowerCase().endsWith(".json"));
if (isJson) {
form.setFieldsValue({ hydrate_on_import: false });
}
}}
/>
<div style={{ marginTop: 8 }}>
<Typography.Text type="secondary">{file ? `已选择文件:${file.name}` : "未选择文件(可在下方粘贴链接)。"}</Typography.Text>
</div>
{fileIsJson ? (
<div style={{ marginTop: 6 }}>
<Typography.Text type="secondary">已检测到 PixivBatchDownloader JSON:将自动导入元数据与标签,补全开关已自动关闭。</Typography.Text>
</div>
) : null}
</Form.Item>
<Form.Item
@@ -174,9 +189,13 @@ export function ImportPage() {
label="导入后立即补全元数据(推荐)"
name="hydrate_on_import"
valuePropName="checked"
extra="随机质量与筛选依赖元数据覆盖率。需要已配置令牌且 worker 正在运行。"
extra={
fileIsJson
? "PixivBatchDownloader JSON 通常已包含元数据/标签(无需令牌)。如需强制补全,请改用纯链接导入。"
: "随机质量与筛选依赖元数据覆盖率。需要已配置令牌且 worker 正在运行。"
}
>
<Switch />
<Switch disabled={fileIsJson} />
</Form.Item>
</Space>
+4 -4
View File
@@ -294,21 +294,21 @@ export function SettingsPage() {
label="使用第三方反向代理(仅图片上游)"
name="image_proxy_use_pixiv_cat"
valuePropName="checked"
extra="开启后:服务端拉取图片时会把 i.pximg.net 替换为 i.pixiv.*(客户端仍访问本站域名,不会暴露第三方域名)。"
extra="开启后:服务端拉取图片时会把 i.pximg.net 替换为 i.pixiv.*(客户端仍访问本站域名,不会暴露第三方域名)。会按访问地区智能选择上游:大陆优先 i.pixiv.re,非大陆默认 i.pixiv.cat。"
>
<Switch />
</Form.Item>
<Form.Item
label="镜像域名"
name="image_proxy_pximg_mirror_host"
extra="可选 i.pixiv.cat / i.pixiv.re / i.pixiv.nl(建议就近选择更快更稳的)。"
extra="可选 i.pixiv.cat / i.pixiv.re / i.pixiv.nl。未显式指定 pximg_mirror_host 时:大陆访问会自动用 i.pixiv.re;非大陆使用这里选择的镜像(默认 i.pixiv.cat)。"
>
<Select
style={{ maxWidth: 360 }}
options={[
{ value: "i.pixiv.cat", label: "i.pixiv.cat(默认)" },
{ value: "i.pixiv.re", label: "i.pixiv.re(大陆备用)" },
{ value: "i.pixiv.nl", label: "i.pixiv.nl(大陆备用)" },
{ value: "i.pixiv.re", label: "i.pixiv.re(大陆优先)" },
{ value: "i.pixiv.nl", label: "i.pixiv.nl(备用)" },
]}
/>
</Form.Item>