Create database.py

database.py

@@ -0,0 +1,88 @@
+# database.py
+import sqlite3
+import datetime
+import uuid
+from typing import List, Dict, Any
+
+DB_FILE = "social_media_platform.db"
+
+# A simple in-memory cache for agent data to avoid constant DB lookups
+AGENT_CACHE: Dict[str, Any] = {}
+
+def initialize_db():
+    conn = sqlite3.connect(DB_FILE)
+    cursor = conn.cursor()
+    # Create agents table with a secret API token
+    cursor.execute("""
+    CREATE TABLE IF NOT EXISTS agents (
+        agent_id INTEGER PRIMARY KEY,
+        name TEXT NOT NULL UNIQUE,
+        persona TEXT NOT NULL,
+        interests TEXT NOT NULL,
+        api_token TEXT NOT NULL UNIQUE
+    )
+    """)
+    # Other tables
+    cursor.execute("CREATE TABLE IF NOT EXISTS posts (post_id INTEGER PRIMARY KEY, agent_id INTEGER, content TEXT, timestamp TEXT)")
+    cursor.execute("CREATE TABLE IF NOT EXISTS comments (comment_id INTEGER PRIMARY KEY, post_id INTEGER, agent_id INTEGER, content TEXT, timestamp TEXT)")
+    cursor.execute("CREATE TABLE IF NOT EXISTS likes (like_id INTEGER PRIMARY KEY, post_id INTEGER, agent_id INTEGER)")
+    conn.commit()
+    conn.close()
+
+def populate_initial_agents():
+    """Adds the 10 starting agents if they don't exist and prints their tokens."""
+    agents_to_create = [
+        {"name": "TechieTom", "persona": "A cynical but brilliant software developer.", "interests": "AI,programming"},
+        {"name": "NatureNina", "persona": "A cheerful nature photographer.", "interests": "hiking,animals"},
+        {"name": "ChefCarlo", "persona": "A passionate Italian chef.", "interests": "pasta,wine"},
+        {"name": "AstroAlex", "persona": "An enthusiastic astronomer.", "interests": "stars,planets"},
+        {"name": "HistoryHank", "persona": "A meticulous historian.", "interests": "battles,empires"},
+        {"name": "Fit Fiona", "persona": "A motivational fitness coach.", "interests": "running,yoga"},
+        {"name": "GamerGabe", "persona": "A competitive esports player.", "interests": "rpgs,strategy"},
+        {"name": "ArtistAnna", "persona": "A thoughtful abstract painter.", "interests": "color theory,canvases"},
+        {"name": "MusoMike", "persona": "A laid-back indie musician.", "interests": "guitar,vinyl"},
+        {"name": "BookwormBella", "persona": "An avid reader of classic literature.", "interests": "novels,poetry"},
+    ]
+    
+    conn = sqlite3.connect(DB_FILE)
+    cursor = conn.cursor()
+    print("\n--- Agent API Tokens ---")
+    for agent_data in agents_to_create:
+        cursor.execute("SELECT api_token FROM agents WHERE name = ?", (agent_data["name"],))
+        result = cursor.fetchone()
+        if result is None:
+            token = str(uuid.uuid4())
+            cursor.execute(
+                "INSERT INTO agents (name, persona, interests, api_token) VALUES (?, ?, ?, ?)",
+                (agent_data["name"], agent_data["persona"], agent_data["interests"], token)
+            )
+            print(f"  {agent_data['name']}: {token}")
+        else:
+            print(f"  {agent_data['name']}: {result[0]} (already exists)")
+    conn.commit()
+    conn.close()
+    print("------------------------\n")
+
+def get_agent_by_token(token: str) -> Dict[str, Any] | None:
+    if token in AGENT_CACHE:
+        return AGENT_CACHE[token]
+    conn = sqlite3.connect(DB_FILE)
+    cursor = conn.cursor()
+    cursor.execute("SELECT agent_id, name FROM agents WHERE api_token = ?", (token,))
+    agent = cursor.fetchone()
+    conn.close()
+    if agent:
+        agent_data = {"agent_id": agent[0], "name": agent[1]}
+        AGENT_CACHE[token] = agent_data
+        return agent_data
+    return None
+
+def get_timeline(limit: int = 20) -> List[Dict[str, Any]]:
+    conn = sqlite3.connect(DB_FILE)
+    # This makes the output rows behave like dictionaries
+    conn.row_factory = sqlite3.Row
+    cursor = conn.cursor()
+    query = """
+    SELECT
+        p.post_id, p.content, p.timestamp,
+        a.agent_id, a