recommendations

Runtime error

App Files Files Community

Pepguy commited on Jun 19, 2025

Commit

f062f34

verified ·

1 Parent(s): 1148777

Update app.py

Browse files

Files changed (1) hide show

app.py +108 -109

app.py CHANGED Viewed

@@ -1,111 +1,110 @@
-# To run this code you need to install the following dependencies:
-# pip install flask google-genai
-import os
-from flask import Flask, request, render_template_string, jsonify
-from google import genai
-from google.genai import types
-app = Flask(__name__)
-HTML = """
-<!DOCTYPE html>
-<html lang="en">
-<head>
-  <meta charset="UTF-8" />
-  <title>Gemini Test</title>
-</head>
-<body style="font-family:sans-serif;padding:2rem;">
-  <h1>Gemini-2.0-Flash Test</h1>
-  <form id="genai-form">
-    <textarea id="prompt" rows="6" cols="60" placeholder="Enter your prompt here"></textarea><br/><br/>
-    <button type="submit">Generate</button>
-  </form>
-  <pre id="output" style="background:#f4f4f4;padding:1rem;margin-top:1rem;"></pre>
-  <script>
-  const form = document.getElementById('genai-form');
-  form.addEventListener('submit', async e => {
-    e.preventDefault();
-    const prompt = document.getElementById('prompt').value.trim();
-    const out = document.getElementById('output');
-    out.textContent = 'Generating…';
-    try {
-      const resp = await fetch('/generate', {
-        method: 'POST',
-        headers: { 'Content-Type': 'application/json' },
-        body: JSON.stringify({ prompt }),
-      });
-      // if you want to debug status codes:
-      if (!resp.ok) {
-        const errText = await resp.text();
-        throw new Error(`Server returned ${resp.status}: ${errText}`);
-      }
-      const data = await resp.json();
-      if (data.error) {
-        // show the backend error message
-        out.textContent = 'Error: ' + data.error;
-      } else if (data.result) {
-        out.textContent = data.result;
-      } else {
-        out.textContent = 'Unexpected response payload';
-      }
-    } catch (err) {
-      console.error(err);
-      out.textContent = 'Fetch error: ' + err.message;
-    }
-  });
-</script>
-</body>
-</html>
-"""
-def generate_from_gemini(prompt: str) -> str:
-    client = genai.Client(
-        api_key="AIzaSyDolbPUZBPUPvQUu-RGktJmvnUpkcEKIYo",
-    )
-    model = "gemini-2.0-flash"
-    contents = [
-        types.Content(
-            role="user",
-            parts=[types.Part.from_text(text=prompt)],
-        )
-    ]
-    config = types.GenerateContentConfig(response_mime_type="text/plain")
-    # Synchronous, non-streaming call:
-    response = client.models.generate_content(
-        model=model,
-        contents=contents,
-        config=config,
-    )
-    # The full text is in response.text
-    return response.text
-@app.route('/')
-def index():
-    return render_template_string(HTML)
-@app.route('/generate', methods=['POST'])
-def gen():
-    data = request.get_json(silent=True) or {}
-    prompt = data.get("prompt", "")
-    app.logger.info(f"Received prompt: {prompt!r}")
-    if not prompt:
-        return jsonify({"error": "No prompt provided"}), 400
     try:
-        output = generate_from_gemini(prompt)
-        app.logger.info("Generation succeeded, length=%d", len(output))
-        return jsonify({"result": output})
-    except Exception as e:
-        app.logger.exception("Generation failed")
-        return jsonify({"error": str(e)}), 500
 if __name__ == "__main__":
-    port = int(os.environ.get("PORT", 7860))
-    app.run(host="0.0.0.0", port=port)

+# server.py
+from typing import List, Dict, Optional, Tuple
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel, Field
+from sklearn.neighbors import KDTree
+import numpy as np
+# Configurable number of scales
+NUM_SCALES = 8
+# Pydantic models
+class Profile(BaseModel):
+    id: str
+    scales: List[float] = Field(..., min_items=NUM_SCALES, max_items=NUM_SCALES)
+class Content(BaseModel):
+    id: str
+    scales: List[float] = Field(..., min_items=NUM_SCALES, max_items=NUM_SCALES)
+    metadata: Optional[Dict] = None
+class Recommendation(BaseModel):
+    content_id: str
+    distance: float
+# Abstract repository interface
+class DataRepository:
+    def add_user(self, user: Profile): ...
+    def add_content(self, content: Content): ...
+    def get_user(self, user_id: str) -> Profile: ...
+    def query_region(self, mins: List[float], maxs: List[float]) -> List[Content]: ...
+# In-memory implementation
+class InMemoryRepo(DataRepository):
+    def __init__(self):
+        self.users: Dict[str, Profile] = {}
+        self.contents: Dict[str, Content] = {}
+    def add_user(self, user: Profile):
+        self.users[user.id] = user
+    def add_content(self, content: Content):
+        self.contents[content.id] = content
+    def get_user(self, user_id: str) -> Profile:
+        user = self.users.get(user_id)
+        if not user:
+            raise ValueError("User not found")
+        return user
+    def query_region(self, mins: List[float], maxs: List[float]) -> List[Content]:
+        # Filter contents by N-dimensional bounding box
+        return [
+            c for c in self.contents.values()
+            if all(mn <= v <= mx for v, mn, mx in zip(c.scales, mins, maxs))
+        ]
+# FastAPI setup
+app = FastAPI()
+repo = InMemoryRepo()
+@app.post("/users", response_model=Profile)
+def create_user(user: Profile):
+    repo.add_user(user)
+    return user
+@app.post("/contents", response_model=Content)
+def create_content(content: Content):
+    repo.add_content(content)
+    return content
+@app.get("/recommendations/{user_id}", response_model=List[Recommendation])
+def recommend(user_id: str, region_radius: float = 0.2, top_k: int = 5):
     try:
+        user = repo.get_user(user_id)
+    except ValueError:
+        raise HTTPException(status_code=404, detail="User not found")
+    # Build bounding region
+    mins = [max(0.0, v - region_radius) for v in user.scales]
+    maxs = [min(1.0, v + region_radius) for v in user.scales]
+    # Query region
+    candidates = repo.query_region(mins, maxs)
+    if not candidates:
+        return []
+    # Build KDTree over candidates to find nearest neighbors
+    mat = np.array([c.scales for c in candidates])
+    tree = KDTree(mat)
+    dist, idx = tree.query(np.array(user.scales).reshape(1, -1), k=min(top_k, len(candidates)))
+    return [
+        Recommendation(content_id=candidates[i].id, distance=float(d))
+        for d, i in zip(dist[0], idx[0])
+    ]
+# Sample test data when run directly
 if __name__ == "__main__":
+    # Install dependencies: fastapi, uvicorn, numpy, scikit-learn, pydantic
+    import uvicorn
+    # Create random test users and content
+    repo.add_user(Profile(id="user1", scales=list(np.random.rand(NUM_SCALES))))
+    for i in range(1, 21):
+        repo.add_content(Content(
+            id=f"content{i}",
+            scales=list(np.random.rand(NUM_SCALES)),
+            metadata={"title": f"Item {i}"}
+        ))
+    uvicorn.run(app, host="0.0.0.0", port=8000)