Something went wrong. Try again.
A lightweight, personal, self-hosted frontend and local caching proxy for pixiv artworks p.ufal.my.id
pixiv python archiver
Something went wrong. Try again.
9.7 kB · 233 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234#!/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`` intothis script. Browser archiving continues to use viewer.php for the equivalentdatabase write, so fetch.py remains independent from SQLite in both workflows."""
import jsonimport osimport sqlite3import sysimport timefrom 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 <artwork_id> | 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:]))