# retoor from datetime import datetime, timedelta, timezone import httpx from devplacepy import stealth from devplacepy import net_guard from devplacepy.database import ( add_news_usage, get_news_usage, get_table, ) from devplacepy.services.base import BaseService, ConfigField from devplacepy.services.openai_gateway.usage import ( accumulate_usage, new_usage_totals, usage_metric_cards, ) from devplacepy.utils import generate_uid, make_combined_slug from devplacepy.services.audit import record as audit from devplacepy.services.seo_meta import schedule_seo_meta from . import _get_ai_key from .constants import ( AI_MODEL_DEFAULT, AI_URL_DEFAULT, FEATURE_MIN_SCORE, FORMAT_INPUT_MAX_CHARS, FORMAT_MAX_TOKENS, FORMAT_OUTPUT_MAX_CHARS, FORMAT_PROMPT_SPEC, GRADE_MAX_TOKENS, GRADE_PROMPT_SPEC, GRADE_THRESHOLD_DEFAULT, GRADING_RULES_DESCRIPTION, IMG_FETCH_TIMEOUT, IMG_PER_ARTICLE, LANDING_MAX, LANDING_MIN_SCORE, LANDING_RECENCY_DAYS, MIN_BODY_CHARS, NEWS_API_URL_DEFAULT, NEWS_SUMMARY, ) from .clean import ( _extract_grade, _strip_md_fence, clean_news_text, effective_score, reliability_reason, ) from .images import ( _fetch_image_candidate, _flag_shared_placeholders, _get_article_images, _primary_image, ) from .models import ArticleGrade, ImageCandidate class NewsService(BaseService): interval_key = "news_service_interval" min_interval = 60 default_enabled = True title = "News" description = NEWS_SUMMARY details = GRADING_RULES_DESCRIPTION config_fields = [ ConfigField( "news_api_url", "News API URL", type="url", default=NEWS_API_URL_DEFAULT, help="Source feed endpoint articles are fetched from.", group="Source", ), ConfigField( "news_ai_url", "AI grading URL", type="url", default=AI_URL_DEFAULT, help="Chat-completions endpoint used to grade each cleaned article.", group="AI grading", ), ConfigField( "news_ai_model", "AI model", type="str", default=AI_MODEL_DEFAULT, help="Model name sent to the grading endpoint.", group="AI grading", ), ConfigField( "news_grade_prompt", "Grading prompt specification", type="text", default=GRADE_PROMPT_SPEC, help=( "The exact rubric sent to the grading model at temperature 0. " "The cleaned Title, Description and Content are appended " "automatically. It must instruct the model to return only a " "single integer from 1 to 10." ), group="AI grading", ), ConfigField( "news_grade_threshold", "Grade threshold (1-10)", type="int", default=GRADE_THRESHOLD_DEFAULT, minimum=1, maximum=10, help=( "Articles whose effective score (AI grade plus the unique-image " "bonus minus the thin-content penalty) reaches this are " "auto-published; below go to draft." ), group="AI grading", ), ConfigField( "news_ai_key", "AI API key", type="password", default="", secret=True, help="Defaults to the NEWS_AI_KEY env var, then the gateway's internal key.", group="AI grading", ), ConfigField( "news_format_enabled", "Reformat content with AI", type="bool", default=True, help=( "When enabled, every valid article is reformatted into clean " "Markdown (paragraphs, headings, lists) after grading." ), group="AI formatting", ), ConfigField( "news_format_prompt", "Formatting prompt specification", type="text", default=FORMAT_PROMPT_SPEC, help=( "The instruction sent to the AI to reformat each cleaned " "article into Markdown. The Title and the article body are " "appended automatically. It must preserve every fact and " "output only the reformatted Markdown body." ), group="AI formatting", ), ] def __init__(self): super().__init__(name="news", interval_seconds=3600) async def run_once(self) -> None: config = self.get_config() api_url = config["news_api_url"] ai_url = config["news_ai_url"] ai_model = config["news_ai_model"] threshold = config["news_grade_threshold"] format_enabled = config["news_format_enabled"] self.log(f"Fetching news from {api_url}") async with stealth.stealth_async_client(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 rejected_count = 0 skipped_count = 0 usage_totals = new_usage_totals() async with net_guard.guarded_async_client( timeout=IMG_FETCH_TIMEOUT ) as image_client: candidates_by_article: dict[str, list[ImageCandidate]] = {} pending: list[tuple[dict, str, ImageCandidate | None]] = [] for article in articles: external_id = article.get("guid", "") if not external_id: continue if external_id in synced_ids: skipped_count += 1 continue article_uid, is_new = self._resolve_uid(news_table, external_id) link = article.get("link", "") candidates: list[ImageCandidate] = [] if link: raw_images = await _get_article_images(link, client) for src in raw_images[:IMG_PER_ARTICLE]: candidates.append( await _fetch_image_candidate(src, image_client) ) candidates_by_article[article_uid] = candidates pending.append((article, article_uid, None)) _flag_shared_placeholders(candidates_by_article) for article, article_uid, _ in pending: candidates = candidates_by_article[article_uid] ai_grade = await self._grade_article( article, ai_url, ai_model, client, usage_totals ) result = self._grade_article_full( article, ai_grade, candidates ) if result.ai_grade is None: failed_count += 1 sync_status = "grading_failed" elif not result.valid: rejected_count += 1 sync_status = f"rejected_quality:{result.reject_reason}" elif result.effective_score < threshold: draft_count += 1 sync_status = "graded" else: sync_status = "graded" published = ( result.valid and result.effective_score >= threshold ) featured = ( published and result.has_unique_image and result.effective_score >= FEATURE_MIN_SCORE ) formatted_content = "" if result.valid and format_enabled: formatted_content = await self._format_article( article, ai_url, ai_model, client, usage_totals ) saved_new = self._store_article( news_table, images_table, article, article_uid, result, published, featured, threshold, candidates, formatted_content, ) if saved_new: new_count += 1 else: updated_count += 1 self._record_sync(sync_table, article["guid"], sync_status) synced_ids.add(article["guid"]) self._apply_landing_selection(news_table, threshold) if usage_totals["calls"]: add_news_usage(usage_totals) self.log( f"AI usage: {usage_totals['calls']} calls, " f"{usage_totals['total_tokens']} tokens, " f"${usage_totals['cost_usd']:.4f}" ) self.log( f"New {new_count}, updated {updated_count}, draft {draft_count}, " f"rejected {rejected_count}, grading failed {failed_count}, " f"skipped {skipped_count}" ) def collect_metrics(self) -> dict: return {"stats": usage_metric_cards(get_news_usage())} def _resolve_uid(self, news_table, external_id: str) -> tuple[str, bool]: existing = news_table.find_one(external_id=external_id) if existing: return existing["uid"], False return generate_uid(), True def _grade_article_full( self, article: dict, ai_grade: int | None, candidates: list[ImageCandidate], ) -> ArticleGrade: title = clean_news_text(article.get("title", "") or "") description = clean_news_text(article.get("description", "") or "") content = clean_news_text(article.get("content", "") or "") body = f"{description} {content}".strip() url = article.get("link", "") or "" has_unique_image = any(not c.is_placeholder for c in candidates) image_url = _primary_image(candidates) if ai_grade is None: return ArticleGrade( ai_grade=None, effective_score=0, has_unique_image=has_unique_image, image_url=image_url, valid=False, reject_reason="grading_failed", candidates=candidates, ) reason = reliability_reason(title, body, url) if reason: return ArticleGrade( ai_grade=ai_grade, effective_score=ai_grade, has_unique_image=has_unique_image, image_url=image_url, valid=False, reject_reason=reason, candidates=candidates, ) body_marginal = len(body) < (MIN_BODY_CHARS * 2) score = effective_score(ai_grade, has_unique_image, body_marginal) return ArticleGrade( ai_grade=ai_grade, effective_score=score, has_unique_image=has_unique_image, image_url=image_url, valid=True, reject_reason="", candidates=candidates, ) def _store_article( self, news_table, images_table, article: dict, article_uid: str, result: ArticleGrade, published: bool, featured: bool, threshold: int, candidates: list[ImageCandidate], formatted_content: str = "", ) -> bool: now = datetime.now(timezone.utc).isoformat() external_id = article.get("guid", "") title = clean_news_text(article.get("title", "") or "")[:500] or "news" description = clean_news_text(article.get("description", "") or "")[:5000] if formatted_content: content = formatted_content[:FORMAT_OUTPUT_MAX_CHARS] else: content = clean_news_text(article.get("content", "") or "")[:10000] status = "published" if published else "draft" existing = news_table.find_one(external_id=external_id) if existing: existing_slug = existing.get("slug", "") featured_locked = existing.get("featured_locked", 0) landing_locked = existing.get("landing_locked", 0) update_row = { "id": existing["id"], "grade": result.effective_score, "ai_grade": result.ai_grade or 0, "status": status, "title": title, "slug": existing_slug or make_combined_slug(title, existing["uid"]), "description": description, "url": article.get("link", ""), "source_name": article.get("feed_name", ""), "content": content, "author": article.get("author", ""), "article_published": article.get("published", ""), "image_url": result.image_url, "has_unique_image": 1 if result.has_unique_image else 0, "synced_at": now, } if not featured_locked: update_row["featured"] = 1 if featured else 0 news_table.update(update_row, ["id"]) images_table.delete(news_uid=existing["uid"]) self._store_images(images_table, existing["uid"], candidates) if status == "published": schedule_seo_meta("news", existing["uid"], regenerate=True) return False slug = make_combined_slug(title, article_uid) news_table.insert( { "uid": article_uid, "slug": slug, "external_id": external_id, "title": title, "description": description, "url": article.get("link", ""), "image_url": result.image_url, "has_unique_image": 1 if result.has_unique_image else 0, "source_name": article.get("feed_name", ""), "grade": result.effective_score, "ai_grade": result.ai_grade or 0, "status": status, "featured": 1 if featured else 0, "featured_locked": 0, "landing_locked": 0, "show_on_landing": 0, "content": content, "author": article.get("author", ""), "article_published": article.get("published", ""), "synced_at": now, "deleted_at": None, "deleted_by": None, } ) self._store_images(images_table, article_uid, candidates) if status == "published": schedule_seo_meta("news", article_uid) audit.record_system( "news.service.ingest", actor_kind="service", target_type="news", target_uid=article_uid, target_label=title, metadata={ "source": article.get("feed_name", ""), "grade": result.effective_score, "ai_grade": result.ai_grade, "unique_image": result.has_unique_image, }, summary=f"news article {title} ingested", links=[audit.target("news", article_uid, title)], ) if result.valid: audit.record_system( "news.service.publish" if published else "news.service.draft", actor_kind="service", target_type="news", target_uid=article_uid, target_label=title, metadata={ "grade": result.effective_score, "threshold": threshold, "featured": featured, }, summary=( f"news article {title} " f"{'auto-published' if published else 'held as draft'}" ), links=[audit.target("news", article_uid, title)], ) else: audit.record_system( "news.service.reject", actor_kind="service", target_type="news", target_uid=article_uid, target_label=title, metadata={"reason": result.reject_reason}, summary=f"news article {title} rejected ({result.reject_reason})", links=[audit.target("news", article_uid, title)], ) return True def _store_images( self, images_table, news_uid: str, candidates: list[ImageCandidate] ) -> None: for candidate in candidates: images_table.insert( { "uid": generate_uid(), "news_uid": news_uid, "url": candidate.url, "alt_text": candidate.alt_text, "phash": candidate.phash, "width": candidate.width, "height": candidate.height, "is_placeholder": 1 if candidate.is_placeholder else 0, "deleted_at": None, "deleted_by": None, } ) def _record_sync(self, sync_table, external_id: str, status: str) -> None: now = datetime.now(timezone.utc).isoformat() existing = sync_table.find_one(external_id=external_id) if existing: sync_table.update( {"id": existing["id"], "status": status, "synced_at": now}, ["id"] ) else: sync_table.insert( { "uid": generate_uid(), "external_id": external_id, "status": status, "synced_at": now, } ) def _apply_landing_selection(self, news_table, threshold: int) -> None: cutoff = ( datetime.now(timezone.utc) - timedelta(days=LANDING_RECENCY_DAYS) ).isoformat() managed = list( news_table.find( deleted_at=None, status="published", featured=1, has_unique_image=1, landing_locked=0, synced_at={">=": cutoff}, order_by=["-grade", "-synced_at"], ) ) chosen: list[str] = [] for article in managed: if ( len(chosen) < LANDING_MAX and article.get("grade", 0) >= LANDING_MIN_SCORE ): chosen.append(article["uid"]) chosen_set = set(chosen) for article in managed: target = 1 if article["uid"] in chosen_set else 0 if article.get("show_on_landing", 0) != target: news_table.update( {"uid": article["uid"], "show_on_landing": target}, ["uid"] ) if target: audit.record_system( "news.service.landing", actor_kind="service", target_type="news", target_uid=article["uid"], target_label=article.get("title"), metadata={"grade": article.get("grade", 0)}, summary=( f"news article {article.get('title')} promoted to landing" ), links=[ audit.target( "news", article["uid"], article.get("title") ) ], ) async def _grade_article( self, article: dict, ai_url: str, ai_model: str, client: httpx.AsyncClient, totals: dict | None = None, ) -> int | None: title = clean_news_text(article.get("title", "") or "")[:500] description = clean_news_text(article.get("description", "") or "")[:1000] content = clean_news_text(article.get("content", "") or "")[:1500] spec = self.get_config().get("news_grade_prompt", "") or GRADE_PROMPT_SPEC prompt = ( f"{spec}\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", "X-App-Reference": "devplace-news-v-1-0-0", } 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() accumulate_usage(totals, resp) 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 async def _format_article( self, article: dict, ai_url: str, ai_model: str, client: httpx.AsyncClient, totals: dict | None = None, ) -> str: title = clean_news_text(article.get("title", "") or "")[:500] description = clean_news_text(article.get("description", "") or "") content = clean_news_text(article.get("content", "") or "") body = f"{description}\n\n{content}".strip() if not body: return "" body = body[:FORMAT_INPUT_MAX_CHARS] spec = self.get_config().get("news_format_prompt", "") or FORMAT_PROMPT_SPEC prompt = f"{spec}\n\nTitle: {title}\n\nArticle:\n{body}" payload = { "model": ai_model, "messages": [{"role": "user", "content": prompt}], "max_tokens": FORMAT_MAX_TOKENS, "temperature": 0.3, } headers = {"Content-Type": "application/json", "X-App-Reference": "devplace-news-v-1-0-0"} 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=60.0 ) if resp.status_code != 200: self.log( f"AI formatting returned {resp.status_code}: {resp.text[:200]}" ) resp.raise_for_status() accumulate_usage(totals, resp) result = resp.json() text = result.get("choices", [{}])[0].get("message", {}).get("content", "") text = _strip_md_fence(text or "") if len(text) < MIN_BODY_CHARS: self.log(f"AI formatting returned too little for: {title[:60]}") return "" return text except Exception as e: self.log(f"AI formatting failed for '{title[:50]}': {e}") return ""