367 lines
12 KiB
Python
367 lines
12 KiB
Python
from quart import Quart, render_template, request, jsonify
|
|
import httpx
|
|
from anilistapi.queries.manga import (
|
|
REQUESTS_QUERY,
|
|
TAGS_QUERY,
|
|
GENRES_QUERY,
|
|
REQUESTED_QUERY,
|
|
)
|
|
from anilistapi.schemas.manga import Manga
|
|
from komconfig import KomConfig
|
|
from komcache import KomCache
|
|
from komgapi import komgapi as KOMGAPI
|
|
from typing import Any, Dict, List
|
|
from limit import limit
|
|
import loguru
|
|
import sys
|
|
from komsuite_nyaapy import Nyaa
|
|
|
|
log = loguru.logger
|
|
log.remove()
|
|
log.add("application.log", rotation="1 week", retention="1 month")
|
|
log.add(sys.stdout)
|
|
app = Quart(__name__)
|
|
|
|
|
|
settings = KomConfig()
|
|
|
|
|
|
# else:
|
|
# cache.create_table(
|
|
# "CREATE TABLE IF NOT EXISTS manga_requests (id INT AUTO_INCREMENT PRIMARY KEY, anilist_id INT NOT NULL, grabbed BOOLEAN DEFAULT 0)"
|
|
|
|
komga = KOMGAPI(
|
|
url=settings.komga.url,
|
|
username=settings.komga.user,
|
|
password=settings.komga.password,
|
|
)
|
|
# komga_series = komga.seriesList()
|
|
|
|
|
|
@limit(limit=1, every=2)
|
|
async def fetch_data(
|
|
inputdata: Dict[str, Any], check_downloads: bool = False
|
|
) -> List[Dict[str, Any]]:
|
|
log.error(f"fetch_data called with data: {inputdata}")
|
|
async with httpx.AsyncClient() as client:
|
|
try:
|
|
# Log the incoming data for debugging
|
|
|
|
# Ensure 'query' key exists in the data dictionary
|
|
if "query" not in inputdata or not inputdata["query"]:
|
|
raise ValueError("Missing or empty 'query' key in data")
|
|
|
|
variables: Dict[str, Any] = {"search": inputdata["query"]}
|
|
if inputdata.get("genres"):
|
|
variables["genres"] = inputdata["genres"]
|
|
if inputdata.get("tags"):
|
|
variables["tags"] = inputdata["tags"]
|
|
if inputdata.get("type"):
|
|
if inputdata["type"] in ["MANGA", "NOVEL"]:
|
|
variables["format"] = inputdata["type"]
|
|
|
|
log.debug(f"GraphQL variables: {variables}")
|
|
|
|
response = await client.post(
|
|
"https://graphql.anilist.co",
|
|
json={
|
|
"query": REQUESTS_QUERY,
|
|
"variables": variables,
|
|
},
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
"User-Agent": "KomPage Searcher API/1.0 (contact: discord-id: 908987973264089139)",
|
|
},
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
log.debug(f"GraphQL response: {data}")
|
|
|
|
results: List[Dict[str, Any]] = []
|
|
cache = KomCache()
|
|
for item in data.get("data", {}).get("Page", {}).get("media", []):
|
|
manga = Manga(**item)
|
|
in_komga = komga.getSeries(
|
|
manga.title.english if manga.title.english else manga.title.romaji
|
|
)
|
|
|
|
requested = cache.fetch_one(
|
|
query="SELECT manga_id, grabbed FROM manga_requests WHERE manga_id = :id",
|
|
args={"id": manga.id},
|
|
)
|
|
komga_request = bool(requested)
|
|
results.append(
|
|
{
|
|
"id": manga.id,
|
|
"title": manga.title.english
|
|
if manga.title.english
|
|
else manga.title.romaji,
|
|
"image": manga.coverImage.get("large")
|
|
if manga.coverImage
|
|
else "https://demofree.sirv.com/nope-not-here.jpg",
|
|
"status": manga.status,
|
|
"type": manga.format,
|
|
"genres": manga.genres or [],
|
|
"tags": [tag.name for tag in (manga.tags or [])],
|
|
"description": manga.description.replace("<br>", "\n")
|
|
if manga.description
|
|
else "No description available",
|
|
"isAdult": manga.isAdult,
|
|
"in_komga": in_komga,
|
|
"requested": komga_request,
|
|
"siteUrl": manga.siteUrl,
|
|
}
|
|
)
|
|
if check_downloads:
|
|
for result in results:
|
|
downloads = Nyaa().search(result.get("title"), 3, 1)
|
|
else:
|
|
downloads = []
|
|
result["download"] = len(downloads) if downloads else 0
|
|
|
|
log.debug(f"Fetched {len(results)} results for query: {inputdata['query']}")
|
|
log.error(f"Results: {results}")
|
|
return results
|
|
except ValueError as ve:
|
|
log.error(f"ValueError in fetch_data: {ve}")
|
|
return []
|
|
except httpx.RequestError as e:
|
|
log.error(f"HTTP request error in fetch_data: {e}")
|
|
return []
|
|
except Exception as e:
|
|
log.error(f"Unexpected error in fetch_data: {e}")
|
|
return []
|
|
|
|
|
|
@limit(limit=1, every=2)
|
|
async def fetch_requested_data(data: list[int]) -> List[Dict[str, Any]]:
|
|
requested_data: list[dict[str, Any]] = []
|
|
|
|
cache = KomCache()
|
|
for manga_id in data:
|
|
async with httpx.AsyncClient() as client:
|
|
try:
|
|
variables: Dict[str, Any] = {
|
|
"search": manga_id,
|
|
"type": "MANGA",
|
|
} # TODO: replace the type once the UI has been updated
|
|
# print(data, variables)
|
|
response = await client.post(
|
|
"https://graphql.anilist.co",
|
|
json={
|
|
"query": REQUESTED_QUERY,
|
|
"variables": variables,
|
|
},
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
# print(data)
|
|
entry = data.get("data", {}).get("Media", {})
|
|
if entry:
|
|
manga = Manga(**entry)
|
|
|
|
in_komga = komga.getSeries(
|
|
manga.title.english
|
|
if manga.title.english
|
|
else manga.title.romaji
|
|
)
|
|
requested = cache.fetch_one(
|
|
query="SELECT grabbed FROM manga_requests WHERE manga_id = :id",
|
|
args={"id": manga.id},
|
|
)
|
|
komga_request = bool(requested)
|
|
|
|
requested_data.append(
|
|
{
|
|
"id": manga.id,
|
|
"title": manga.title.english
|
|
if manga.title.english
|
|
else manga.title.romaji,
|
|
"image": manga.coverImage.get("large")
|
|
if manga.coverImage
|
|
else None,
|
|
"status": manga.status,
|
|
"type": manga.format,
|
|
"genres": manga.genres or [],
|
|
"tags": [tag.name for tag in (manga.tags or [])],
|
|
"description": manga.description.replace("<br>", "\n")
|
|
if manga.description
|
|
else None,
|
|
"isAdult": manga.isAdult,
|
|
"in_komga": bool(in_komga),
|
|
"requested": bool(komga_request),
|
|
}
|
|
)
|
|
log.debug(
|
|
requested_data,
|
|
)
|
|
|
|
except httpx.RequestError as e:
|
|
print(f"An error occurred while requesting data: {e}")
|
|
requested_data.append({"id": manga_id})
|
|
except Exception as e:
|
|
print(f"Unexpected error: {e}")
|
|
requested_data.append({"id": manga_id})
|
|
return requested_data
|
|
|
|
|
|
@app.route("/api/genres")
|
|
async def get_genres():
|
|
async with httpx.AsyncClient() as client:
|
|
try:
|
|
response = await client.post(
|
|
"https://graphql.anilist.co",
|
|
json={"query": GENRES_QUERY},
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
results = data.get("data", {}).get("genres", [])
|
|
return jsonify(results)
|
|
except Exception as e:
|
|
print(f"Error fetching genres: {e}")
|
|
return jsonify([])
|
|
|
|
|
|
@app.route("/api/tags")
|
|
async def get_tags():
|
|
async with httpx.AsyncClient() as client:
|
|
try:
|
|
response = await client.post(
|
|
"https://graphql.anilist.co",
|
|
json={"query": TAGS_QUERY},
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
results = data.get("data", {}).get("tags", [])
|
|
return jsonify(results)
|
|
except Exception as e:
|
|
print(f"Error fetching genres: {e}")
|
|
return jsonify([])
|
|
|
|
|
|
@app.route("/search", methods=["POST"])
|
|
async def search():
|
|
data = await request.get_json()
|
|
print(data)
|
|
results = await fetch_data(data)
|
|
if not results:
|
|
return jsonify({"error": "No results found"}), 404
|
|
return jsonify(results)
|
|
|
|
|
|
@app.route("/api/search", methods=["GET"])
|
|
async def api_search():
|
|
"""
|
|
API endpoint for searching manga.
|
|
|
|
Parameters:
|
|
- q: Query string to search for
|
|
|
|
Returns:
|
|
- JSON array of matching manga entries
|
|
"""
|
|
try:
|
|
query = request.args.get("q")
|
|
|
|
if not query:
|
|
return jsonify({"error": "Missing required parameter: q"}), 400
|
|
|
|
# Build search parameters
|
|
search_data: Dict[str, Any] = {"query": query}
|
|
|
|
# Add optional filters if provided
|
|
|
|
log.info(f"API search request: {search_data}")
|
|
|
|
# Fetch results using existing fetch_data function
|
|
results = await fetch_data(search_data, check_downloads=False)
|
|
|
|
if not results:
|
|
return jsonify({"results": [], "count": 0}), 404
|
|
|
|
return jsonify({"results": results, "count": len(results)})
|
|
|
|
except Exception as e:
|
|
log.error(f"Error in API search: {e}")
|
|
return jsonify({"error": "An unexpected error occurred"}), 500
|
|
|
|
|
|
@app.route("/", methods=["GET"])
|
|
async def index():
|
|
return await render_template("index.html") # , komga_series=komga_series)
|
|
|
|
|
|
@app.route("/request", methods=["POST"])
|
|
async def log_request():
|
|
data = await request.get_json()
|
|
log.debug(f"Received request data: {data}")
|
|
item = data.get("item")
|
|
if item:
|
|
# data = await fetch_requested_data([item])
|
|
if not data:
|
|
return jsonify({"status": "failed", "message": "Item not found"}), 404
|
|
# data = data[0]
|
|
manga_id = item.get("id")
|
|
title = item.get("title")
|
|
image_url = item.get("image")
|
|
log.debug(
|
|
f"Logging request for item: {item}, title: {title}, image_url: {image_url}, manga_id: {manga_id}"
|
|
)
|
|
asynccache = KomCache()
|
|
asynccache.insert(
|
|
f"INSERT INTO manga_requests (manga_id, title, image) VALUES (:manga_id, :title, :image)",
|
|
args={"manga_id": manga_id, "title": title, "image": image_url},
|
|
)
|
|
return jsonify({"status": "success"})
|
|
return jsonify({"status": "failed"}), 400
|
|
|
|
|
|
@app.route("/requests", methods=["GET"])
|
|
async def requests_page():
|
|
cache = KomCache()
|
|
requests = (
|
|
cache.fetch_all(
|
|
query="SELECT manga_id, title, image, grabbed FROM manga_requests"
|
|
)
|
|
or []
|
|
)
|
|
entries: List[Dict[str, Any]] = []
|
|
req_ids = [req[0] for req in requests]
|
|
if req_ids:
|
|
entries = [
|
|
{"manga_id": req[0], "title": req[1], "image": req[2], "grabbed": req[3]}
|
|
for req in requests
|
|
]
|
|
else:
|
|
entries = []
|
|
log.debug(f"Fetched {len(entries)} requests from the database")
|
|
log.debug(f"Requests entries: {entries}")
|
|
return await render_template("requests.html", requests=entries)
|
|
|
|
@app.route("/delete", methods=["POST"])
|
|
async def delete_request():
|
|
# Delete a request from the database. ID is sent after the /delete endpoint, so: /delete/<id>
|
|
data = await request.get_json()
|
|
log.debug(f"Received delete request data: {data}")
|
|
|
|
if data:
|
|
title = data.get("title")
|
|
asynccache = KomCache()
|
|
asynccache.delete(
|
|
"DELETE FROM manga_requests WHERE title = :title", args={"title": title}
|
|
)
|
|
return jsonify({"status": "success"})
|
|
return jsonify({"status": "failed"}), 400
|
|
|
|
if __name__ == "__main__":
|
|
log.info("Starting Komga Manga Grabber API")
|
|
# use hypercorn to run the app
|
|
import uvicorn
|
|
|
|
uvicorn.run(app, host="0.0.0.0", port=5001)
|
|
# run in test mode
|