KomSuite/KomPage
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects 1 Releases 10 Wiki Activity

update files, add release workflow

Browse Source
This commit is contained in:
WorldTeacher
2025-05-23 15:45:10 +02:00
parent 1d0f293e19
commit 10b47d2f13
9 changed files with 372 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
src/app.py
View File

@@ -12,7 +12,14 @@ 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__)
cache = KomCache()
@@ -32,19 +39,28 @@ komga_series = komga.seriesList()
@limit(90, 60)
async def fetch_data(
data: Dict[str, Any], query: str = REQUESTS_QUERY
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:
variables: Dict[str, Any] = {"search": data["query"]}
if data.get("genres"):
variables["genres"] = data["genres"]
if data.get("tags"):
variables["tags"] = data["tags"]
if data.get("type"):
if data["type"] in ["MANGA", "NOVEL"]:
variables["format"] = data["type"]
print(data["query"], variables)
# 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={
@@ -55,6 +71,7 @@ async def fetch_data(
response.raise_for_status()
data = response.json()
log.debug(f"GraphQL response: {data}")
results: List[Dict[str, Any]] = []
for item in data.get("data", {}).get("Page", {}).get("media", []):
@@ -67,7 +84,6 @@ async def fetch_data(
args=(manga.id,),
)
komga_request = bool(requested)
results.append(
{
"id": manga.id,
@@ -87,18 +103,30 @@ async def fetch_data(
"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:
print(f"An error occurred while requesting data: {e}")
log.error(f"HTTP request error in fetch_data: {e}")
return []
except Exception as e:
print(f"Unexpected error: {e}")
log.error(f"Unexpected error in fetch_data: {e}")
return []
@limit(90, 60)
async def fetch_requested_data(data: list[int]) -> List[Dict[str, Any]]:
requested_data: list[dict[str, Any]] = []
for manga_id in data:
@@ -130,7 +158,7 @@ async def fetch_requested_data(data: list[int]) -> List[Dict[str, Any]]:
else manga.title.romaji
)
requested = cache.fetch_one(
query="SELECT manga_id, grabbed FROM manga_requests WHERE manga_id = ?",
query="SELECT grabbed FROM manga_requests WHERE manga_id = ?",
args=(manga.id,),
)
komga_request = bool(requested)
@@ -143,19 +171,22 @@ async def fetch_requested_data(data: list[int]) -> List[Dict[str, Any]]:
else manga.title.romaji,
"image": manga.coverImage.get("large")
if manga.coverImage
else "https://demofree.sirv.com/nope-not-here.jpg",
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 "No description available",
else None,
"isAdult": manga.isAdult,
"in_komga": in_komga,
"requested": komga_request,
"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}")
@@ -211,6 +242,43 @@ async def search():
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)
@@ -241,9 +309,11 @@ async def requests_page():
entries = await fetch_requested_data(req_ids)
else:
entries = []
print(entries)
data = []
for entry in entries:
data.append(dict(entry))
return await render_template("requests.html", requests=entries)
return await render_template("requests.html", requests=data)
if __name__ == "__main__":