devplacepy/content.py at 5fe9b4f1ed283d8a036173c7955f636c5a3cf88b

 # retoor <retoor@molodetz.nl>
 from collections import Counter
 from devplacepy.cache import TTLCache
 from .core import db, get_table, or_
 _daily_topic_cache = TTLCache(ttl=60, max_size=1)
 _trending_cache = TTLCache(ttl=15, max_size=1)
 def resolve_by_slug(table, slug, include_deleted=False):
     has_soft_delete = table.has_column("deleted_at")
     flt = {} if include_deleted or not has_soft_delete else {"deleted_at": None}
     entry = table.find_one(slug=slug, **flt)
     if not entry:
         entry = table.find_one(uid=slug, **flt)
     return entry
 def resolve_object_url(target_type: str, target_uid: str) -> str:
     if target_type == "post":
         post = resolve_by_slug(get_table("posts"), target_uid)
         return f"/posts/{post['slug'] or post['uid']}" if post else "/feed"
     if target_type == "project":
         project = resolve_by_slug(get_table("projects"), target_uid)
         return (
             f"/projects/{project['slug'] or project['uid']}" if project else "/projects"
         )
     if target_type == "news":
         article = resolve_by_slug(get_table("news"), target_uid)
         if article:
             return f"/news/{article.get('slug', '') or article['uid']}"
         return "/news"
     if target_type == "issue":
         return f"/issues?highlight={target_uid}"
     if target_type == "gist":
         gist = resolve_by_slug(get_table("gists"), target_uid)
         return f"/gists/{gist['slug'] or gist['uid']}" if gist else "/gists"
     if target_type == "comment":
         comment = get_table("comments").find_one(uid=target_uid, deleted_at=None)
         if not comment:
             return "/feed"
         parent_url = resolve_object_url(
             comment.get("target_type", "post"),
             comment.get("target_uid") or comment.get("post_uid", ""),
         )
         return f"{parent_url}#comment-{target_uid}"
     if target_type == "award":
         award = resolve_by_slug(get_table("awards"), target_uid)
         if award:
             receiver = get_table("users").find_one(uid=award.get("receiver_uid", ""))
             if receiver:
                 return f"/profile/{receiver['username']}?tab=awards#award-{award.get('slug', '')}"
     return "/feed"
 def get_uids_by_username_match(search, limit=200):
     term = (search or "").strip()
     if not term or "users" not in db.tables:
         return []
     rows = db.query(
         "SELECT uid FROM users WHERE username LIKE :q LIMIT :limit",
         q=f"%{term}%",
         limit=limit,
     )
     return [row["uid"] for row in rows]
 def text_search_clause(
     table, search, fields=("title", "description"), author_field=None
 ):
     if not search or not search.strip() or not table.exists:
         return None
     columns = table.table.columns
     like = f"%{search.strip()}%"
     matches = [columns[field].ilike(like) for field in fields if field in columns]
     if author_field and author_field in columns:
         author_uids = get_uids_by_username_match(search)
         if author_uids:
             matches.append(columns[author_field].in_(author_uids))
     return or_(*matches) if matches else None
 def get_daily_topic():
     cached = _daily_topic_cache.get("topic")
     if cached is not None:
         return cached
     topic = _load_daily_topic()
     _daily_topic_cache.set("topic", topic)
     return topic
 def _load_daily_topic():
     if "news" in db.tables:
         article = db["news"].find_one(
             status="published", deleted_at=None, order_by=["-synced_at"]
         )
         if article:
             desc = (article.get("description") or "")[:200] or (
                 article.get("content") or ""
             )[:200]
             return {
                 "title": article.get("title", ""),
                 "summary": desc,
                 "slug": article.get("slug", ""),
                 "url": article.get("url", ""),
                 "image_url": article.get("image_url", ""),
             }
     return {
         "title": "Welcome to DevPlace",
         "summary": "Stay tuned for the latest dev news.",
     }
 def get_featured_news(limit=5):
     if "news" not in db.tables:
         return []
     from devplacepy.utils import time_ago
     rows = list(
         db["news"].find(
             show_on_landing=1, deleted_at=None, order_by=["-synced_at"], _limit=limit
         )
     )
     articles = []
     for article in rows:
         summary = (article.get("description") or "")[:120] or (
             article.get("content") or ""
         )[:120]
         articles.append(
             {
                 "title": article.get("title", ""),
                 "summary": summary,
                 "slug": article.get("slug", ""),
                 "url": article.get("url", ""),
                 "source_name": article.get("source_name", ""),
                 "featured": article.get("featured", 0),
                 "image_url": article.get("image_url", "") or "",
                 "time_ago": time_ago(article["synced_at"])
                 if article.get("synced_at")
                 else "",
             }
         )
     return articles
 def get_trending_topics(limit: int = 6) -> list[dict]:
     cached = _trending_cache.get("topics")
     if cached is not None:
         return cached[:limit]
     if "posts" not in db.tables or "topic" not in db["posts"].columns:
         return []
     rows = db.query(
         "SELECT topic FROM posts WHERE deleted_at IS NULL "
         "AND topic IS NOT NULL AND topic != '' "
         "ORDER BY created_at DESC LIMIT 200"
     )
     counter: Counter[str] = Counter()
     for row in rows:
         topic = (row["topic"] or "").strip()
         if topic:
             counter[topic] += 1
     topics = [{"topic": t, "count": c} for t, c in counter.most_common(limit)]
     _trending_cache.set("topics", topics)
     return topics