devplacepy/ai_context.py at ef1c914e23d2a2e71d07d638e95d71366d534be9

 # retoor <retoor@molodetz.nl>
 import logging
 from datetime import datetime, timezone
 from devplacepy.database import (
     db,
     get_follow_counts,
     get_table,
     get_user_post_count,
     get_user_rank,
 )
 logger = logging.getLogger(__name__)
 MAX_BIO = 280
 MAX_EXCERPT = 400
 MAX_SOURCE = 1500
 MAX_THREAD_MESSAGES = 6
 MAX_MESSAGE = 240
 TARGET_TABLES = {
     "post": "posts",
     "project": "projects",
     "gist": "gists",
     "news": "news",
 }
 def _truncate(text: str, limit: int) -> str:
     text = " ".join((text or "").split())
     if len(text) <= limit:
         return text
     return text[:limit].rstrip() + "..."
 def _today() -> str:
     return datetime.now(timezone.utc).strftime("%d/%m/%Y")
 def _author_block(user_uid: str) -> str:
     user = get_table("users").find_one(uid=user_uid)
     if not user:
         return ""
     stats = [
         f"role {user.get('role') or 'Member'}",
         f"level {user.get('level', 1)}",
         f"{user.get('stars', 0)} stars",
     ]
     try:
         stats.append(f"{get_user_post_count(user_uid)} posts")
     except Exception:
         pass
     try:
         rank = get_user_rank(user_uid)
         if rank:
             stats.append(f"rank #{rank}")
     except Exception:
         pass
     try:
         follows = get_follow_counts(user_uid)
         stats.append(f"{follows.get('followers', 0)} followers")
     except Exception:
         pass
     joined = (user.get("created_at") or "")[:10]
     if joined:
         stats.append(f"member since {joined}")
     line = f"Author: {user.get('username', '')} ({', '.join(stats)})."
     bio = _truncate(user.get("bio") or "", MAX_BIO)
     if bio:
         line += f' Bio: "{bio}".'
     return line
 def _target_excerpt(target_type: str, target_uid: str) -> str:
     table = TARGET_TABLES.get(target_type)
     if not table or table not in db.tables or not target_uid:
         return ""
     row = get_table(table).find_one(uid=target_uid)
     if not row:
         return ""
     title = (row.get("title") or "").strip()
     body = _truncate(row.get("content") or row.get("description") or "", MAX_EXCERPT)
     label = f'the {target_type} "{title}"' if title else f"a {target_type}"
     text = f"This is a comment on {label}."
     if body:
         text += f' Excerpt: "{body}".'
     return text
 def _comment_block(row: dict) -> str:
     target_type = row.get("target_type") or "post"
     target_uid = row.get("target_uid") or row.get("post_uid") or ""
     text = _target_excerpt(target_type, target_uid)
     parent_uid = row.get("parent_uid")
     if parent_uid and "comments" in db.tables:
         parent = get_table("comments").find_one(uid=parent_uid)
         if parent:
             text += f' In reply to: "{_truncate(parent.get("content") or "", MAX_EXCERPT)}".'
     return text
 def _post_block(row: dict) -> str:
     topic = row.get("topic") or "general"
     text = f"This is a post (topic: {topic})."
     project_uid = row.get("project_uid")
     if project_uid and "projects" in db.tables:
         project = get_table("projects").find_one(uid=project_uid)
         if project and project.get("title"):
             text += f' Attached to the project "{project["title"]}".'
     return text
 def _item_block(table: str, row: dict) -> str:
     kind = "project" if table == "projects" else "gist"
     title = (row.get("title") or "").strip()
     text = f'This is a {kind}' + (f' titled "{title}".' if title else ".")
     description = _truncate(row.get("description") or "", MAX_EXCERPT)
     if description:
         text += f' Description: "{description}".'
     if table == "gists":
         language = row.get("language") or "plaintext"
         source = _truncate(row.get("source_code") or "", MAX_SOURCE)
         text += f" Language: {language}."
         if source:
             text += f" Source code (reference only, do not include unless asked): {source}"
     return text
 def _message_block(row: dict) -> str:
     sender_uid = row.get("sender_uid")
     receiver_uid = row.get("receiver_uid")
     receiver = get_table("users").find_one(uid=receiver_uid) if receiver_uid else None
     receiver_name = (receiver or {}).get("username") or "the recipient"
     text = f"This is a direct message to {receiver_name}."
     if "messages" not in db.tables or not sender_uid or not receiver_uid:
         return text
     rows = list(
         db.query(
             "SELECT sender_uid, content FROM messages "
             "WHERE ((sender_uid = :a AND receiver_uid = :b) "
             "OR (sender_uid = :b AND receiver_uid = :a)) "
             "AND uid != :current AND deleted_at IS NULL "
             "ORDER BY id DESC LIMIT :lim",
             a=sender_uid,
             b=receiver_uid,
             current=row.get("uid") or "",
             lim=MAX_THREAD_MESSAGES,
         )
     )
     if not rows:
         return text
     lines = []
     for entry in reversed(rows):
         who = "you" if entry["sender_uid"] == sender_uid else receiver_name
         lines.append(f"{who}: {_truncate(entry['content'] or '', MAX_MESSAGE)}")
     return text + " Recent conversation:\n" + "\n".join(lines)
 def _location_block(table: str, row: dict) -> str:
     if table == "comments":
         return _comment_block(row)
     if table == "posts":
         return _post_block(row)
     if table in ("projects", "gists"):
         return _item_block(table, row)
     if table == "messages":
         return _message_block(row)
     return ""
 def build_context(table: str, uid: str, row: dict, user_uid: str) -> str:
     parts = [f"Today is {_today()} on the DevPlace developer network."]
     try:
         author = _author_block(user_uid)
         if author:
             parts.append(author)
     except Exception as exc:
         logger.warning("ai context author block failed: %s", exc)
     try:
         location = _location_block(table, row)
         if location:
             parts.append(location)
     except Exception as exc:
         logger.warning("ai context location block failed: %s", exc)
     return "\n".join(parts)