import logging
import os
import re
from datetime import datetime, timezone
import httpx
from devplacepy.database import get_table, get_setting
from devplacepy.services.base import BaseService
from devplacepy.utils import generate_uid, make_combined_slug, strip_html
logger = logging.getLogger(__name__)
NEWS_API_URL_DEFAULT = "https://news.app.molodetz.nl/api"
AI_URL_DEFAULT = "https://openai.app.molodetz.nl/v1/chat/completions"
AI_MODEL_DEFAULT = "molodetz"
GRADE_THRESHOLD_DEFAULT = 7
GRADE_MAX_TOKENS = 2000
def _get_ai_key() -> str:
key = os.environ.get("NEWS_AI_KEY")
if key:
return key
key = get_setting("news_ai_key", "")
if key:
return key
key = os.environ.get("OPENROUTER_API_KEY")
if key:
return key
key = os.environ.get("OPENAI_API_KEY")
if key:
return key
return ""
def _extract_grade(text: str) -> int | None:
match = re.search(r"\d+", text.strip())
if match:
val = int(match.group())
if 1 <= val <= 10:
return val
return None
async def _get_article_images(url: str, client: httpx.AsyncClient) -> list[dict]:
try:
resp = await client.get(url, timeout=10.0)
resp.raise_for_status()
html = resp.text
pattern = re.compile(r'<img[^>]+src=["\']([^"\']+)["\']', re.IGNORECASE)
matches = pattern.findall(html)
images = []
seen = set()
for src in matches:
if src in seen:
continue
seen.add(src)
if src.startswith("http") and not any(ext in src.lower() for ext in [".svg", ".ico"]):
images.append({"url": src, "alt_text": ""})
return images[:10]
except Exception:
return []
class NewsService(BaseService):
def __init__(self):
super().__init__(name="news", interval_seconds=3600)
async def run_once(self) -> None:
api_url = get_setting("news_api_url", NEWS_API_URL_DEFAULT)
ai_url = get_setting("news_ai_url", AI_URL_DEFAULT)
ai_model = get_setting("news_ai_model", AI_MODEL_DEFAULT)
threshold = int(get_setting("news_grade_threshold", str(GRADE_THRESHOLD_DEFAULT)))
self.log(f"Fetching news from {api_url}")
async with httpx.AsyncClient(timeout=30.0) as client:
try:
resp = await client.get(api_url)
resp.raise_for_status()
data = resp.json()
except Exception as e:
self.log(f"Failed to fetch news API: {e}")
return
articles = data.get("articles", [])
self.log(f"Received {len(articles)} articles")
news_table = get_table("news")
images_table = get_table("news_images")
sync_table = get_table("news_sync")
synced_ids = set()
for entry in sync_table.find():
synced_ids.add(entry["external_id"])
new_count = 0
updated_count = 0
draft_count = 0
failed_count = 0
skipped_count = 0
for article in articles:
external_id = article.get("guid", "")
if not external_id:
continue
if external_id in synced_ids:
skipped_count += 1
continue
grade = await self._grade_article(article, ai_url, ai_model, client)
now = datetime.now(timezone.utc).isoformat()
if grade is None:
grade_val = 0
auto_published = False
failed_count += 1
sync_status = "grading_failed"
elif grade < threshold:
grade_val = grade
auto_published = False
draft_count += 1
sync_status = "graded"
else:
grade_val = grade
auto_published = True
sync_status = "graded"
is_published = "published" if auto_published else "draft"
existing = news_table.find_one(external_id=external_id)
if existing:
existing_slug = existing.get("slug", "")
news_table.update({
"id": existing["id"],
"grade": grade_val,
"status": is_published,
"title": article.get("title", ""),
"slug": existing_slug or make_combined_slug(article.get("title", "") or "news", existing["uid"]),
"description": strip_html(article.get("description", "") or "")[:5000],
"url": article.get("link", ""),
"source_name": article.get("feed_name", ""),
"content": strip_html(article.get("content", "") or "")[:10000],
"author": article.get("author", ""),
"article_published": article.get("published", ""),
"synced_at": now,
}, ["id"])
updated_count += 1
images_table.delete(news_uid=existing["uid"])
article_uid = existing["uid"]
else:
article_uid = generate_uid()
article_slug = make_combined_slug(article.get("title", "") or "news", article_uid)
news_table.insert({
"uid": article_uid,
"slug": article_slug,
"external_id": external_id,
"title": article.get("title", ""),
"description": strip_html(article.get("description", "") or "")[:5000],
"url": article.get("link", ""),
"image_url": "",
"source_name": article.get("feed_name", ""),
"grade": grade_val,
"status": is_published,
"content": strip_html(article.get("content", "") or "")[:10000],
"author": article.get("author", ""),
"article_published": article.get("published", ""),
"synced_at": now,
})
new_count += 1
existing_sync = sync_table.find_one(external_id=external_id)
if existing_sync:
sync_table.update({
"id": existing_sync["id"],
"status": sync_status,
"synced_at": now,
}, ["id"])
else:
sync_table.insert({
"uid": generate_uid(),
"external_id": external_id,
"status": sync_status,
"synced_at": now,
})
synced_ids.add(external_id)
link = article.get("link", "")
if link:
fresh_images = await _get_article_images(link, client)
for img in fresh_images:
images_table.insert({
"uid": generate_uid(),
"news_uid": article_uid,
"url": img["url"],
"alt_text": img.get("alt_text", ""),
})
self.log(f"New {new_count}, updated {updated_count}, draft {draft_count}, "
f"grading failed {failed_count}, skipped {skipped_count}")
async def _grade_article(self, article: dict, ai_url: str, ai_model: str, client: httpx.AsyncClient) -> int | None:
title = (article.get("title", "") or "")[:500]
description = strip_html(article.get("description", "") or "")[:1000]
content = strip_html(article.get("content", "") or "")[:1500]
prompt = (
"You are an expert content strategist and editor for DevPlace, a "
"server-rendered social network for software developers. Decide how "
"strongly a given article should be published, expressed as a single "
"grade.\n\n"
"Site context:\n"
"- Audience: software developers, hobbyists and professionals, "
"interested in software development, AI agents, open-source tools, "
"systems programming, and self-hosting.\n"
"- Mission: deliver deep technical insight, critical analysis, and "
"practical value; build authority and reader trust; avoid low-effort, "
"rehashed, or clickbait content.\n"
"- Tone: technical, practical, nuanced, skeptical of hype; no "
"sensationalism or unverified claims.\n\n"
"Evaluate the article against these weighted criteria and weigh them "
"internally:\n"
"1. Relevance & alignment to the developer niche and audience needs "
"(20%).\n"
"2. Quality & accuracy: depth, originality, factual correctness, "
"sourcing, technical correctness, clear writing (25%).\n"
"3. Engagement & readability: strong title/intro/flow, useful "
"examples, appropriate length (15%).\n"
"4. Originality & uniqueness: new perspective, data, or analysis; not "
"duplicated generic web content (15%).\n"
"5. SEO & discoverability: keyword opportunity without stuffing, "
"shareable (10%).\n"
"6. Ethics, legality & risk: no misinformation, plagiarism, or "
"brand-safety issues; controversial topics handled with nuance and "
"evidence (10%).\n"
"7. Strategic fit for the site (5%).\n\n"
"Convert your overall judgment to a single integer grade from 1 to "
"10:\n"
"- 9-10: excellent, publish as-is.\n"
"- 7-8: good, worth publishing.\n"
"- 5-6: marginal, weak fit or quality.\n"
"- 1-4: reject, low quality or off-topic.\n\n"
"Return ONLY a single integer from 1 to 10, nothing else.\n\n"
f"Title: {title}\n"
f"Description: {description}\n"
f"Content: {content}"
)
payload = {
"model": ai_model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": GRADE_MAX_TOKENS,
"temperature": 0.0,
}
headers = {
"Content-Type": "application/json",
}
ai_key = _get_ai_key()
if ai_key:
headers["Authorization"] = f"Bearer {ai_key}"
try:
resp = await client.post(ai_url, json=payload, headers=headers, timeout=15.0)
if resp.status_code != 200:
self.log(f"AI grading returned {resp.status_code}: {resp.text[:200]}")
resp.raise_for_status()
result = resp.json()
text = result.get("choices", [{}])[0].get("message", {}).get("content", "")
if not text:
self.log(f"AI grading returned empty content for: {title[:60]}")
return None
grade = _extract_grade(text)
if grade is None:
self.log(f"AI grading returned unparseable response: {text[:100]}")
return grade
except Exception as e:
self.log(f"AI grading failed for '{title[:50]}': {e}")
return None
Reference in New Issue View Git Blame Copy Permalink