diff --git a/core/orchestrator.py b/core/orchestrator.py index 28c7f08..e60b67a 100644 --- a/core/orchestrator.py +++ b/core/orchestrator.py @@ -358,14 +358,21 @@ Respond with a JSON array of strings, each a compelling article title.""" def _score_and_assign(raw_topics: list[str]) -> list[dict]: - """Score topics and assign to verticals using LLM.""" + """Score topics and assign to verticals using LLM, boosted by learning data.""" if not raw_topics: return [] + # Phase 0: Get learning insights from live sites + learning_insights = _get_learning_insights() + # Deduplicate first unique = list(dict.fromkeys(raw_topics))[:50] - prompt = f"""You are a content strategist. Score and categorize these {len(unique)} topics. + insights_text = "" + if learning_insights: + insights_text = f"\n\nLEARNING DATA — content that performs well on our sites:\n{json.dumps(learning_insights, indent=2)}\n\nUse this to boost composite_score for topics similar to what our audience already reads. Topics matching high-performing patterns get +10 to composite_score." + + prompt = f"""You are a content strategist. Score and categorize these {len(unique)} topics.{insights_text} Topics: {json.dumps(unique)} @@ -378,7 +385,7 @@ For each topic, return: - "competition_score": 0-100 (how many competing articles exist) - "freshness_score": 0-100 (how new/urgent) - "evergreen_score": 0-100 (will this be relevant in 5 years) -- "composite_score": overall value score 0-100 (higher = publish now) +- "composite_score": overall value score 0-100 (higher = publish now) — apply learning boosts here Vertical assignment rules: - AI/ML topics → ai @@ -395,13 +402,100 @@ Respond with a JSON array of objects. No markdown, no explanation.""" try: result = ollama_json(prompt, model="qwen3.5:4b", temperature=0.3) if isinstance(result, list): - return result + # Apply algorithmic boost on top of LLM scores + return _apply_learning_boost(result, learning_insights) return [] except Exception as e: log.warning(f"Topic scoring failed: {e}") return [] +def _get_learning_insights() -> dict: + """Query all 8 live sites for their top-performing content patterns.""" + insights = {} + for vertical, vinfo in VERTICALS.items(): + ct_ip = vinfo.get("ip") + if not ct_ip: + continue + try: + r = requests.get(f"http://{ct_ip}:5000/api/stats", timeout=5) + if r.status_code == 200: + data = r.json() + popular = data.get("popular", []) + if popular: + # Extract keyword patterns from popular articles + all_keywords = [] + for art in popular: + kw_str = art.get("keywords", "[]") + try: + kws = json.loads(kw_str) if isinstance(kw_str, str) else kw_str + all_keywords.extend(kws) + except (json.JSONDecodeError, TypeError): + pass + + # Most frequent keywords = winning topics + from collections import Counter + kw_counts = Counter(all_keywords) + + insights[vertical] = { + "top_keywords": [kw for kw, _ in kw_counts.most_common(8)], + "top_articles": [a.get("title", "")[:80] for a in popular[:3]], + "avg_word_count": sum(a.get("word_count", 0) for a in popular) // max(len(popular), 1), + "total_articles": data.get("total_articles", 0), + } + except Exception: + pass + + # Also check orchestrator's own performance_learning DB + try: + db = sqlite3.connect(str(DB_PATH)) + db.row_factory = sqlite3.Row + for vertical in VERTICALS: + row = db.execute( + "SELECT * FROM performance_learning WHERE vertical=? ORDER BY updated_at DESC LIMIT 1", + (vertical,) + ).fetchone() + if row: + if vertical not in insights: + insights[vertical] = {} + try: + insights[vertical]["stored_patterns"] = json.loads(row["top_patterns"]) + except (json.JSONDecodeError, TypeError): + pass + db.close() + except Exception: + pass + + return insights if insights else {} + + +def _apply_learning_boost(scored: list[dict], insights: dict) -> list[dict]: + """Boost composite scores for topics matching winning patterns.""" + if not insights: + return scored + + for topic in scored: + vertical = topic.get("vertical", "") + title = topic.get("title", "").lower() + vin = insights.get(vertical, {}) + top_kws = vin.get("top_keywords", []) + + # Count keyword matches between topic title and winning keywords + matches = sum(1 for kw in top_kws if kw.lower() in title) + boost = min(matches * 8, 25) # Up to 25-point boost + + # Boost for matching the optimal word count range (signals topic depth fits) + if vin.get("avg_word_count", 0) > 0: + boost += 3 # Minor boost for having any data + + if boost > 0: + old_score = topic.get("composite_score", 50) + topic["composite_score"] = min(old_score + boost, 100) + topic["learning_boost"] = boost + + return scored + + def _deduplicate_and_rank(scored: list[dict]) -> list[dict]: """Remove near-duplicates and rank by composite score.""" seen_titles = set() diff --git a/sites/_engine/app.py b/sites/_engine/app.py index 889ef0d..d21a29c 100644 --- a/sites/_engine/app.py +++ b/sites/_engine/app.py @@ -508,6 +508,62 @@ def api_stats(): }) +@app.route("/api/learn", methods=["POST"]) +def api_learn(): + """Nightly learning: analyze performance and update internal models.""" + db = get_db() + + # Aggregate keyword performance + rows = db.execute(""" + SELECT a.keywords, COUNT(p.id) as views + FROM articles a + LEFT JOIN pageviews p ON a.id = p.article_id + WHERE a.status = 'published' + GROUP BY a.id + ORDER BY views DESC + LIMIT 20 + """).fetchall() + + keyword_views = {} + for row in rows: + try: + kws = json.loads(row["keywords"]) if isinstance(row["keywords"], str) else (row["keywords"] or []) + except (json.JSONDecodeError, TypeError): + kws = [] + for kw in kws: + keyword_views[kw] = keyword_views.get(kw, 0) + (row["views"] or 0) + + top_keywords = sorted(keyword_views.items(), key=lambda x: x[1], reverse=True)[:15] + + # Best performing word count range + wc_row = db.execute(""" + SELECT AVG(a.word_count) as avg_wc, AVG(a.reading_time) as avg_rt + FROM articles a + LEFT JOIN pageviews p ON a.id = p.article_id + WHERE a.status = 'published' + GROUP BY a.id + HAVING COUNT(p.id) > 0 + ORDER BY COUNT(p.id) DESC + LIMIT 10 + """).fetchone() + + # Store learning + db.execute(""" + INSERT OR REPLACE INTO learning (metric, value, recorded_at) + VALUES ('top_keywords', ?, datetime('now')) + """, (json.dumps(top_keywords),)) + + db.commit() + + return jsonify({ + "status": "learned", + "top_keywords": [{"keyword": kw, "views": v} for kw, v in top_keywords[:10]], + "optimal_word_count": round(wc_row["avg_wc"]) if wc_row and wc_row["avg_wc"] else None, + "optimal_reading_time": round(wc_row["avg_rt"]) if wc_row and wc_row["avg_rt"] else None, + "total_articles_analyzed": db.execute("SELECT COUNT(*) FROM articles WHERE status='published'").fetchone()[0], + }) + + @app.route("/health") def health(): db = get_db()