|
|
from fastapi import FastAPI, Query, BackgroundTasks |
|
|
from fastapi.responses import FileResponse |
|
|
import aiohttp |
|
|
import asyncio |
|
|
import time |
|
|
import tempfile |
|
|
import zipfile |
|
|
import os |
|
|
from pydantic import BaseModel |
|
|
from typing import List, Dict |
|
|
|
|
|
|
|
|
img_base = 'https://i.pximg.net/img-original/img/' |
|
|
|
|
|
class pixifModel(BaseModel): |
|
|
post_ids: List[int] |
|
|
phpsessid: str |
|
|
|
|
|
class PixifDownloadModel(BaseModel): |
|
|
posts: Dict[str, str] |
|
|
|
|
|
class PixifZipModel(BaseModel): |
|
|
d: Dict[str, str] |
|
|
|
|
|
headers = { |
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:126.0) Gecko/20100101 Firefox/126.0", |
|
|
'referer': 'https://www.pixiv.net/', |
|
|
} |
|
|
|
|
|
app = FastAPI() |
|
|
|
|
|
async def fetch_page(session, url): |
|
|
async with session.get(url) as response: |
|
|
data = await response.json() |
|
|
return data |
|
|
|
|
|
def build_cookies(phpsessid: str) -> Dict[str, str]: |
|
|
return {"PHPSESSID": phpsessid} |
|
|
|
|
|
def base26_time(): |
|
|
x='' |
|
|
n=int(time.time()*100) |
|
|
while n: |
|
|
x=chr(97+n%26)+x |
|
|
n//=26 |
|
|
|
|
|
return x |
|
|
|
|
|
def determine_exif_type(metadata): |
|
|
if metadata is None: |
|
|
return None |
|
|
elif metadata == b'TitleAI generated image': |
|
|
return "novelai" |
|
|
elif metadata.startswith(b"parameter"): |
|
|
return "sd" |
|
|
elif b'{"' in metadata: |
|
|
return "comfy" |
|
|
elif b"Dig" in metadata: |
|
|
return "mj" |
|
|
elif metadata.startswith(b"SoftwareCelsys"): |
|
|
return "celsys" |
|
|
else: |
|
|
return "photoshop" |
|
|
|
|
|
async def get_exif(url, session): |
|
|
start_range = 0 |
|
|
end_range = 512 |
|
|
|
|
|
headers = { |
|
|
"Referer": "https://www.pixiv.net/", |
|
|
"Range": f"bytes={start_range}-{end_range}" |
|
|
} |
|
|
|
|
|
async with session.get(url, headers=headers) as response: |
|
|
data = await response.read() |
|
|
return parse_png_metadata(data) |
|
|
|
|
|
def parse_png_metadata(data): |
|
|
index = 8 |
|
|
|
|
|
while index < len(data): |
|
|
if index + 8 > len(data): |
|
|
break |
|
|
chunk_len = int.from_bytes(data[index:index+4], 'big') |
|
|
chunk_type = data[index+4:index+8].decode('ascii') |
|
|
index += 8 |
|
|
|
|
|
if chunk_type in ['tEXt', 'iTXt']: |
|
|
content = data[index:index+chunk_len] |
|
|
if chunk_type == 'tEXt': |
|
|
return content.replace(b'\0', b'') |
|
|
elif chunk_type == 'iTXt': |
|
|
return content.strip() |
|
|
|
|
|
index += chunk_len + 4 |
|
|
return None |
|
|
|
|
|
async def process_post(post_id, session, semaphore): |
|
|
async with semaphore: |
|
|
try: |
|
|
data = await fetch_page(session, f"https://www.pixiv.net/ajax/illust/{post_id}/pages") |
|
|
image_urls = [page['urls']['original'] for page in data['body'] if 'png' in page['urls']['original']] |
|
|
|
|
|
initial_offsets = [1, 5, 5, 10, 10, 10] |
|
|
|
|
|
chunks = [] |
|
|
start = 0 |
|
|
for offset in initial_offsets: |
|
|
end = start + offset |
|
|
if end > len(image_urls): |
|
|
end = len(image_urls) |
|
|
chunks.append((start, end)) |
|
|
start = end |
|
|
|
|
|
while start < len(image_urls): |
|
|
end = min(start + 10, len(image_urls)) |
|
|
chunks.append((start, end)) |
|
|
start = end |
|
|
|
|
|
for s, e in chunks: |
|
|
chunk_tasks = [get_exif(image_urls[i], session) for i in range(s, e)] |
|
|
results = await asyncio.gather(*chunk_tasks) |
|
|
|
|
|
for image_url, metadata in zip(image_urls[s:e], results): |
|
|
exif_type = determine_exif_type(metadata) |
|
|
if exif_type not in ['photoshop', 'celsys', None]: |
|
|
return post_id, image_url |
|
|
|
|
|
return post_id, None |
|
|
except Exception: |
|
|
return post_id, None |
|
|
|
|
|
async def fetch_image_bytes(session, url, post_id, semaphore): |
|
|
async with semaphore: |
|
|
async with session.get(url) as response: |
|
|
return post_id, await response.read() |
|
|
|
|
|
@app.get("/allimages") |
|
|
async def all_images( |
|
|
only_first = Query("0", description="Only fetch the first image of each post."), |
|
|
post_ids: List[int] = Query(..., alias='post_ids'), |
|
|
phpsessid: str = Query(..., description="Pixiv PHPSESSID value.") |
|
|
): |
|
|
cookies = build_cookies(phpsessid) |
|
|
async with aiohttp.ClientSession(cookies=cookies, headers=headers) as session: |
|
|
tasks = [] |
|
|
for pid in post_ids: |
|
|
tasks.append(fetch_page(session, f"https://www.pixiv.net/ajax/illust/{pid}/pages")) |
|
|
|
|
|
results = await asyncio.gather(*tasks) |
|
|
|
|
|
all_image_urls = [] |
|
|
for data in results: |
|
|
if "body" in data: |
|
|
for page in data["body"]: |
|
|
if "urls" in page and "original" in page["urls"]: |
|
|
all_image_urls.append(page["urls"]["original"]) |
|
|
|
|
|
if only_first != "0": |
|
|
break |
|
|
|
|
|
return {"image_urls": all_image_urls} |
|
|
|
|
|
@app.get("/allimage") |
|
|
async def all_image( |
|
|
post_id: int = Query(..., description="The post ID to fetch all images from."), |
|
|
phpsessid: str = Query(..., description="Pixiv PHPSESSID value.") |
|
|
): |
|
|
cookies = build_cookies(phpsessid) |
|
|
async with aiohttp.ClientSession(cookies=cookies, headers=headers) as session: |
|
|
try: |
|
|
data = await fetch_page(session, f"https://www.pixiv.net/ajax/illust/{post_id}/pages") |
|
|
image_urls = [page['urls']['original'] for page in data['body']] |
|
|
return {"post_id": post_id, "image_urls": image_urls} |
|
|
except Exception as e: |
|
|
return {"error": str(e)} |
|
|
|
|
|
@app.post("/pixif") |
|
|
async def pixif( |
|
|
items: pixifModel |
|
|
): |
|
|
post_ids = items.post_ids |
|
|
semaphore = asyncio.Semaphore(100) |
|
|
cookies = build_cookies(items.phpsessid) |
|
|
|
|
|
async with aiohttp.ClientSession(cookies=cookies, headers=headers) as session: |
|
|
tasks = [process_post(post_id, session, semaphore) for post_id in post_ids] |
|
|
results = await asyncio.gather(*tasks) |
|
|
|
|
|
image_exifs = {post_id: image_url.replace(img_base, '', 1) for post_id, image_url in results if image_url} |
|
|
|
|
|
return image_exifs |
|
|
|
|
|
@app.post("/pixif_zip") |
|
|
async def pixif_zip(items: PixifZipModel, background_tasks: BackgroundTasks): |
|
|
downloads = items.d |
|
|
if not downloads: |
|
|
return {"detail": "No downloads requested."} |
|
|
|
|
|
tmp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".zip") |
|
|
tmp_path = tmp_file.name |
|
|
tmp_file.close() |
|
|
|
|
|
connector = aiohttp.TCPConnector(limit=20) |
|
|
semaphore = asyncio.Semaphore(20) |
|
|
async with aiohttp.ClientSession(headers=headers, connector=connector) as session: |
|
|
tasks = [] |
|
|
for post_id, url in downloads.items(): |
|
|
full_url = url if url.startswith("http") else img_base + url |
|
|
tasks.append(asyncio.create_task(fetch_image_bytes(session, full_url, post_id, semaphore))) |
|
|
|
|
|
with zipfile.ZipFile(tmp_path, "w", compression=zipfile.ZIP_DEFLATED) as zf: |
|
|
for task in asyncio.as_completed(tasks): |
|
|
try: |
|
|
post_id, data = await task |
|
|
except Exception: |
|
|
continue |
|
|
if data: |
|
|
zf.writestr(f"{post_id}.png", data) |
|
|
|
|
|
background_tasks.add_task(os.remove, tmp_path) |
|
|
filename = f"pixif_{base26_time()}.zip" |
|
|
return FileResponse(tmp_path, media_type="application/zip", filename=filename) |
|
|
|
|
|
@app.get("/") |
|
|
async def read_root(): |
|
|
return {"message": "Hello, World!"} |
|
|
|
|
|
if __name__ == "__main__": |
|
|
import uvicorn |
|
|
uvicorn.run(app, host="127.0.0.1", port=7860) |
|
|
|
