#!/usr/bin/env python3 """Validate fetched artwork JSON and upsert it into localpixiv's SQLite DB. The CLI wrapper pipes the single JSON document produced by ``fetch.py`` into this script. Browser archiving continues to use viewer.php for the equivalent database write, so fetch.py remains independent from SQLite in both workflows. """ import json import os import sqlite3 import sys import time from urllib.parse import urlparse # insert.py is invoked from the project wrapper, but absolute paths make direct # execution equally safe from any working directory. SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) PROJECT_DIR = os.path.abspath(os.path.join(SCRIPT_DIR, "..", "..")) DB_FILE = os.path.join(PROJECT_DIR, "db", "cache.sqlite") def ensure_schema(connection): """Create the artwork table and add columns used by current localpixiv.""" # New installations receive the complete current table in one statement. connection.execute( """ CREATE TABLE IF NOT EXISTS artworks ( id TEXT PRIMARY KEY, title TEXT, artist_name TEXT, artist_avatar TEXT, tags TEXT, translated_tags TEXT, photos TEXT, local_photos TEXT, page_count INTEGER, is_r18 INTEGER DEFAULT 0, is_ai INTEGER DEFAULT 0, created_at INTEGER, create_date TEXT, sanity_level INTEGER DEFAULT 2, view_count INTEGER DEFAULT 0, weekly_view_count INTEGER DEFAULT 0 ) """ ) # Older personal databases may predate translated tags or view counters. # Apply only additive migrations so archived rows are never rebuilt or lost. existing = { row[1] for row in connection.execute("PRAGMA table_info(artworks)") } additions = { "translated_tags": "TEXT", "create_date": "TEXT", "sanity_level": "INTEGER DEFAULT 2", "view_count": "INTEGER DEFAULT 0", "weekly_view_count": "INTEGER DEFAULT 0", } for column, definition in additions.items(): if column not in existing: connection.execute( f"ALTER TABLE artworks ADD COLUMN {column} {definition}" ) def require_string_list(value, field): """Return a validated list containing only strings.""" # Tags and URL manifests cross a process boundary as JSON; explicit type # checks prevent accidental serialization of dictionaries or scalar values. if not isinstance(value, list) or not all(isinstance(item, str) for item in value): raise ValueError(f"{field} must be an array of strings.") return value def validate_metadata(metadata): """Normalize one fetch.py result and reject unsafe or incomplete shapes.""" # fetch.py represents local exceptions and pixiv API failures with an error # object. Such an envelope must never become an artwork database row. if not isinstance(metadata, dict): raise ValueError("Fetch output must be a JSON object.") if metadata.get("error"): raise ValueError(f"Fetch failed: {metadata['error']}") artwork_id = str(metadata.get("id", "")) if not artwork_id.isdigit(): raise ValueError("Artwork ID must be numeric.") # photos is the authoritative ordered page manifest used later by repair. # Restrict it to absolute HTTP(S) URLs before storing it persistently. photos = require_string_list(metadata.get("photos"), "photos") if not photos: raise ValueError("photos must contain at least one source URL.") for url in photos: parsed = urlparse(url) if parsed.scheme not in ("http", "https") or not parsed.netloc: raise ValueError("photos contains an invalid source URL.") # A None slot is allowed for a failed page so metadata can still be retained # and repaired later. Any non-null path must be the exact canonical location, # preventing piped JSON from injecting arbitrary filesystem paths. local_photos = metadata.get("local_photos") if not isinstance(local_photos, list) or len(local_photos) != len(photos): raise ValueError("local_photos must match the number of source pages.") for page_index, path in enumerate(local_photos): expected = f"cache/images/{artwork_id}/{page_index}.webp" if path is not None and path != expected: raise ValueError(f"Unexpected local path for page {page_index}.") # Normalize optional fields into the same shapes used by viewer.php's upsert. tags = require_string_list(metadata.get("tags", []), "tags") translated_tags = require_string_list( metadata.get("translated_tags", []), "translated_tags" ) return { "id": artwork_id, "title": str(metadata.get("title", "")), "artist_name": str(metadata.get("artist_name", "")), "artist_avatar": str(metadata.get("artist_avatar", "")), "tags": tags, "translated_tags": translated_tags, "photos": photos, "local_photos": local_photos, "page_count": len(photos), "is_r18": 1 if metadata.get("is_r18") else 0, "is_ai": 1 if metadata.get("is_ai") else 0, "create_date": str(metadata.get("create_date", "")), "sanity_level": int(metadata.get("sanity_level", 2)), } def upsert_artwork(metadata, database_file=DB_FILE): """Insert new metadata while preserving archive time and view counters.""" artwork = validate_metadata(metadata) # The directory creation only supports first-run CLI use; normal instances # already have db/ created by installation or the web front controller. os.makedirs(os.path.dirname(database_file), exist_ok=True) connection = sqlite3.connect(database_file, timeout=15) try: # Match the web application's WAL and lock-wait behavior so CLI archives # can coexist with gallery reads and view-counter writes. connection.execute("PRAGMA busy_timeout = 5000") connection.execute("PRAGMA journal_mode = WAL") with connection: ensure_schema(connection) # created_at and both view counters are intentionally absent from the # conflict update. Re-fetching metadata must not make an old artwork # appear newly archived or erase accumulated views. connection.execute( """ INSERT INTO artworks ( id, title, artist_name, artist_avatar, tags, translated_tags, photos, local_photos, page_count, is_r18, is_ai, created_at, create_date, sanity_level ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET title = excluded.title, artist_name = excluded.artist_name, artist_avatar = excluded.artist_avatar, tags = excluded.tags, translated_tags = excluded.translated_tags, photos = excluded.photos, local_photos = excluded.local_photos, page_count = excluded.page_count, is_r18 = excluded.is_r18, is_ai = excluded.is_ai, create_date = excluded.create_date, sanity_level = excluded.sanity_level """, ( artwork["id"], artwork["title"], artwork["artist_name"], artwork["artist_avatar"], ",".join(artwork["tags"]), ",".join(artwork["translated_tags"]), json.dumps(artwork["photos"], ensure_ascii=False), json.dumps(artwork["local_photos"], ensure_ascii=False), artwork["page_count"], artwork["is_r18"], artwork["is_ai"], int(time.time()), artwork["create_date"], artwork["sanity_level"], ), ) finally: connection.close() return artwork def main(arguments): """Read one JSON document from stdin and persist it transactionally.""" if arguments: print("Usage: fetch.py | insert.py", file=sys.stderr) return 1 # Reading the complete stdin stream ensures that fetch.py contributes exactly # one JSON document; progress and network diagnostics belong on stderr. try: raw_input = sys.stdin.read() if not raw_input.strip(): raise ValueError("No fetch JSON was provided on standard input.") metadata = json.loads(raw_input) artwork = upsert_artwork(metadata) except (ValueError, TypeError, json.JSONDecodeError, sqlite3.Error, OSError) as error: print(f"Unable to insert artwork: {error}", file=sys.stderr) return 1 print( f"Archived {artwork['id']}: {artwork['title']} " f"({artwork['page_count']} page(s))." ) # Preserve incomplete metadata for the repair command, but return a distinct # failure status so wrappers such as bookmark.py can retry the archive. missing_pages = [ index for index, path in enumerate(artwork["local_photos"]) if path is None ] if missing_pages: joined = ", ".join(str(index) for index in missing_pages) print( f"Image cache is incomplete; missing page(s): {joined}.", file=sys.stderr, ) return 2 return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))