Merge pull request #234 from Willy-JL/fix/indexer-qol

Indexer: Bugfixes and Last Updated date improvements part 1

Author: WillyJL

indexer/cache.py

@@ -17,7 +17,7 @@
 SHORT_TTL = dt.timedelta(days=2).total_seconds()
 LAST_CHANGE_ELIGIBLE_FIELDS = (
     "name",
-    "thread_version",
+    "version",
     "developer",
     "type",
     "status",
@@ -29,7 +29,10 @@
     "tags",
     "unknown_tags",
     "image_url",
+    "previews_urls",
     "downloads",
+    "reviews_total",
+    "reviews",
     "INDEX_ERROR",
 )
 
@@ -38,13 +41,13 @@
 locks_lock = asyncio.Lock()
 locks: dict[asyncio.Lock] = {}
 
-CACHE_KEYWORDS = (
-    LAST_CACHED := "LAST_CACHED",
+LAST_CACHED = "LAST_CACHED"
+EXPIRE_TIME = "EXPIRE_TIME"
+INDEX_ERROR = "INDEX_ERROR"
+INTERNAL_KEYWORDS = (
     CACHED_WITH := "CACHED_WITH",
     LAST_CHANGE := "LAST_CHANGE",
-    EXPIRE_TIME := "EXPIRE_TIME",
     HASHED_META := "HASHED_META",
-    INDEX_ERROR := "INDEX_ERROR",
 )
 NAME_FORMAT = "thread:{id}"
 
@@ -96,28 +99,19 @@ async def get_thread(id: int) -> dict[str, str]:
 
     thread = await redis.hgetall(name)
 
-    # Don't return thread data (there might be some) if an error flag is active
-    if thread.get(INDEX_ERROR):
-        return {INDEX_ERROR: thread[INDEX_ERROR]}
-
     # Remove internal fields from response
-    for key in CACHE_KEYWORDS:
+    for key in INTERNAL_KEYWORDS:
         if key in thread:
             del thread[key]
     return thread
 
 
 async def _is_thread_cache_outdated(id: int, name: str) -> bool:
-    last_cached, cached_with, expire_time = await redis.hmget(
-        name, (LAST_CACHED, CACHED_WITH, EXPIRE_TIME)
-    )
+    last_cached, expire_time = await redis.hmget(name, (LAST_CACHED, EXPIRE_TIME))
     if last_cached and not expire_time:
         expire_time = int(last_cached) + CACHE_TTL
-    return (
-        not last_cached  # Never cached
-        or time.time() >= int(expire_time)  # Cache expired
-        # or cached_with != meta.version  # Cached on different version
-    )
+    # Never cached or cache expired
+    return not last_cached or time.time() >= int(expire_time)
 
 
 async def _maybe_update_thread_cache(id: int, name: str) -> None:
@@ -148,11 +142,6 @@ async def _update_thread_cache(id: int, name: str) -> None:
             INDEX_ERROR: result.error_flag,
             EXPIRE_TIME: int(now + result.retry_delay),
         }
-        if result is f95zone.ERROR_THREAD_MISSING:
-            # F95zone responded but thread is missing, remove any previous cache
-            await redis.delete(name)
-            if last_change := old_fields.get(LAST_CHANGE):
-                new_fields[LAST_CHANGE] = last_change
         # Consider new error as a change
         if old_fields.get(INDEX_ERROR) != new_fields.get(INDEX_ERROR):
             new_fields[LAST_CHANGE] = int(now)

indexer/f95zone.py

@@ -38,7 +38,7 @@
     b"<p>Automated backups are currently executing. During this time, the site will be unavailable</p>",
     b"<title>F95Zone :: Scheduled Maintenance</title>",
     b'<script src="https://static.f95zone.to/assets/SamF95/ErrorPage',
-    b'<div class="blockMessage"><p>Please check back in 10 mins</p></div>'
+    b'<div class="blockMessage"><p>Please check back in 10 mins</p></div>',
 )
 
 logger = logging.getLogger(__name__)

indexer/scraper.py

@@ -127,7 +127,12 @@ async def thread(id: int) -> dict[str, str] | f95zone.IndexerError | None:
                         parser.attachment(preview_url)
                         for preview_url in update["screens"]
                     ] or ret.previews_urls
-                    ret.last_updated = parser.datestamp(update["ts"])
+                    last_promoted = parser.datestamp(update["ts"])
+                    if (
+                        ret.last_updated > time.time()  # Only if thread has a typo
+                        or last_promoted > ret.last_updated  # Or it's outdated
+                    ):
+                        ret.last_updated = last_promoted
                     break
             else:  # Didn't break
                 continue
@@ -141,7 +146,7 @@ async def thread(id: int) -> dict[str, str] | f95zone.IndexerError | None:
         async with f95zone.RATELIMIT:
             try:
                 async with f95zone.session.get(
-                    thread_url + "/br-reviews",
+                    thread_url + "/br-reviews/",
                     cookies=f95zone.cookies,
                 ) as req:
                     if req.status == 429 and retries > 1:
@@ -159,17 +164,22 @@ async def thread(id: int) -> dict[str, str] | f95zone.IndexerError | None:
     if index_error := f95zone.check_error(res, logger):
         return index_error
 
-    if not str(req.real_url).endswith("br-reviews"):
+    if not str(req.real_url).rstrip("/").endswith("br-reviews"):
         # Some threads have reviews disabled
         reviews = parser.ParsedReviews(total=0, items=[])
     else:
         reviews = await loop.run_in_executor(None, parser.reviews, res)
         if isinstance(reviews, parser.ParserError):
 
-            if reviews.message == "Thread structure missing" and req.status in (403, 404):
+            if reviews.message == "Thread structure missing" and req.status in (
+                403,
+                404,
+            ):
                 return f95zone.ERROR_THREAD_MISSING
 
-            logger.error(f"Thread {id} reviews parsing failed: {reviews.message}\n{reviews.dump}")
+            logger.error(
+                f"Thread {id} reviews parsing failed: {reviews.message}\n{reviews.dump}"
+            )
             return f95zone.ERROR_PARSING_FAILED
 
         reviews.items = [dataclasses.asdict(review) for review in reviews.items]

indexer/watcher.py

@@ -81,7 +81,9 @@ async def watch_updates():
                         try:
                             updates = json.loads(res)
                         except Exception:
-                            raise Exception(f"Latest updates returned invalid JSON: {res}")
+                            raise Exception(
+                                f"Latest updates returned invalid JSON: {res}"
+                            )
                         if index_error := f95zone.check_error(updates, logger):
                             raise Exception(index_error)
 
@@ -94,7 +96,9 @@ async def watch_updates():
                         for update in updates["msg"]["data"]:
                             name = cache.NAME_FORMAT.format(id=update["thread_id"])
                             names.append(name)
-                            cached_data.hmget(name, "version", cache.HASHED_META)
+                            cached_data.hmget(
+                                name, "version", cache.HASHED_META, cache.LAST_CACHED
+                            )
                             version = update["version"]
                             if version == "Unknown":
                                 version = None
@@ -114,18 +118,19 @@ async def watch_updates():
                         cached_data = await cached_data.execute()
 
                         assert len(names) == len(current_data) == len(cached_data)
-                        for name, (version, meta), (cached_version, cached_meta) in zip(
-                            names, current_data, cached_data
-                        ):
-                            if cached_version is None:
+                        for (
+                            name,
+                            (version, meta),
+                            (cached_version, cached_meta, last_cached),
+                        ) in zip(names, current_data, cached_data):
+                            if cached_version is None or not last_cached:
                                 continue
 
                             version_outdated = version and version != cached_version
                             meta_outdated = meta != cached_meta
 
                             if version_outdated or meta_outdated:
-                                # Delete version too to avoid watch_versions() picking it up as mismatch
-                                invalidate_cache.hdel(name, cache.LAST_CACHED, "version")
+                                invalidate_cache.hdel(name, cache.LAST_CACHED)
                                 invalidate_cache.hset(name, cache.HASHED_META, meta)
                                 logger.info(
                                     f"Updates: Invalidating cache for {name}"
@@ -175,16 +180,18 @@ async def watch_versions():
             async with asyncio.timeout(f95zone.TIMEOUT.total):
                 logger.info("Poll versions start")
 
-                names = [n async for n in cache.redis.scan_iter("thread:*", 10000, "hash")]
+                names = [
+                    n async for n in cache.redis.scan_iter("thread:*", 10000, "hash")
+                ]
                 invalidate_cache = cache.redis.pipeline()
 
                 for names_chunk in chunks(names, WATCH_VERSIONS_CHUNK_SIZE):
 
-                    cached_versions = cache.redis.pipeline()
+                    cached_data = cache.redis.pipeline()
                     csv = ""
                     ids = []
                     for name in names_chunk:
-                        cached_versions.hget(name, "version")
+                        cached_data.hmget(name, "version", cache.LAST_CACHED)
                         id = name.split(":")[1]
                         csv += f"{id},"
                         ids.append(id)
@@ -195,8 +202,8 @@ async def watch_versions():
                             f95zone.VERCHK_URL.format(threads=csv),
                         ) as req:
                             # Await together for efficiency
-                            res, cached_versions = await asyncio.gather(
-                                req.read(), cached_versions.execute()
+                            res, cached_data = await asyncio.gather(
+                                req.read(), cached_data.execute()
                             )
                     except Exception as exc:
                         if index_error := f95zone.check_error(exc, logger):
@@ -210,24 +217,28 @@ async def watch_versions():
                         versions = json.loads(res)
                     except Exception:
                         raise Exception(f"Versions API returned invalid JSON: {res}")
-                    if versions.get("msg") in ("Missing threads data", "Thread not found"):
+                    if versions.get("msg") in (
+                        "Missing threads data",
+                        "Thread not found",
+                    ):
                         versions["status"] = "ok"
                         versions["msg"] = {}
                     if index_error := f95zone.check_error(versions, logger):
                         raise Exception(index_error)
                     versions = versions["msg"]
 
-                    assert len(names_chunk) == len(ids) == len(cached_versions)
-                    for name, id, cached_version in zip(names_chunk, ids, cached_versions):
-                        if cached_version is None:
+                    assert len(names_chunk) == len(ids) == len(cached_data)
+                    for name, id, (cached_version, last_cached) in zip(
+                        names_chunk, ids, cached_data
+                    ):
+                        if cached_version is None or not last_cached:
                             continue
                         version = versions.get(id)
                         if not version or version == "Unknown":
                             continue
 
                         if version != cached_version:
-                            # Delete version too to avoid ending up here again
-                            invalidate_cache.hdel(name, cache.LAST_CACHED, "version")
+                            invalidate_cache.hdel(name, cache.LAST_CACHED)
                             logger.warning(
                                 f"Versions: Invalidating cache for {name}"
                                 f" ({cached_version!r} -> {version!r})"
@@ -236,7 +247,9 @@ async def watch_versions():
                 if len(invalidate_cache):
                     result = await invalidate_cache.execute()
                     invalidated = sum(ret != "0" for ret in result)
-                    logger.warning(f"Versions: Invalidated cache for {invalidated} threads")
+                    logger.warning(
+                        f"Versions: Invalidated cache for {invalidated} threads"
+                    )
 
                 logger.info("Poll versions done")