# retoor <retoor@molodetz.nl>
import asyncio
import json
import logging
from devplacepy.database import (
add_seo_usage,
get_seo_usage,
get_table,
has_fresh_seo_metadata,
internal_gateway_key,
upsert_seo_metadata,
)
from devplacepy.seo_meta_text import (
DESCRIPTION_MAX,
TITLE_MAX,
clamp_generated,
plain_seo_defaults,
)
from devplacepy.services.ai_context import build_context
from devplacepy.services.base import ConfigField
from devplacepy.services.correction import gateway_complete, new_usage_totals
from devplacepy.services.jobs import queue
from devplacepy.services.jobs.base import JobService
from devplacepy.services.openai_gateway.usage import usage_metric_cards
from devplacepy.services.seo_meta import (
BODY_FIELD,
TITLE_FIELD,
TYPE_TABLES,
load_target_row,
schedule_seo_meta,
)
logger = logging.getLogger(__name__)
GENERATION_TIMEOUT_SECONDS = 30.0
SYSTEM_PROMPT = (
"You are an SEO metadata engine for a developer social network. From the content "
"below produce search-optimized metadata. Return ONLY a strict JSON object with the "
'keys "seo_title", "seo_description" and "seo_keywords" and nothing else. Rules: '
f"seo_title is at most {TITLE_MAX} characters, front-loads the primary keyword, uses "
"a single hyphen as separator, and contains no markdown. seo_description is at most "
f"{DESCRIPTION_MAX} characters, places the key message within the first 120 "
"characters, reads naturally, and contains no markdown. seo_keywords is a short "
"honest comma-separated list of 5 to 8 lowercase terms, never keyword-stuffed. Use a "
"plain hyphen, never an em-dash."
)
class SeoMetaService(JobService):
kind = "seo_meta"
title = "SEO Metadata"
description = (
"Generates clean SEO title, description and keywords for every published post, "
"project, gist, news article and issue off the request path, and meters its own "
"AI spend. Until the AI value is ready a plain-content default is served so the "
"fields are never empty. The seo_metadata row is permanent; only the job tracking "
"row is swept on retention."
)
def __init__(self):
super().__init__(name="seo_meta", interval_seconds=10)
self.config_fields = list(self.config_fields) + [
ConfigField(
"seo_meta_backfill_enabled",
"Backfill existing content",
type="bool",
default=True,
help="Generate metadata for already-published content lacking it.",
group="SEO metadata",
),
ConfigField(
"seo_meta_backfill_batch",
"Backfill items per tick",
type="int",
default=5,
minimum=1,
maximum=100,
help="How many published items to enqueue for backfill each tick.",
group="SEO metadata",
),
]
async def process(self, job: dict) -> dict:
payload = job["payload"]
target_type = payload.get("target_type", "")
target_uid = str(payload.get("target_uid", ""))
row = self._load_row(target_type, target_uid)
title = (row or {}).get(TITLE_FIELD.get(target_type, "title")) or ""
body = (row or {}).get(BODY_FIELD.get(target_type, "content")) or ""
defaults = plain_seo_defaults(title, body)
if not row:
upsert_seo_metadata(
target_type,
target_uid,
defaults["title"],
defaults["description"],
defaults["keywords"],
"failed",
"plain",
)
self._audit(target_type, target_uid, "plain", title, result="failure")
return {"target_type": target_type, "target_uid": target_uid, "source": "plain"}
totals = new_usage_totals()
result = await asyncio.to_thread(
self._generate, target_type, target_uid, row, title, body, defaults, totals
)
if totals["calls"]:
add_seo_usage(totals)
upsert_seo_metadata(
target_type,
target_uid,
result["title"],
result["description"],
result["keywords"],
"ready" if result["source"] == "ai" else "failed",
result["source"],
)
self._audit(
target_type,
target_uid,
result["source"],
result["title"],
result="success" if result["source"] == "ai" else "failure",
)
return {
"target_type": target_type,
"target_uid": target_uid,
"source": result["source"],
"title_len": len(result["title"]),
"desc_len": len(result["description"]),
}
def _generate(
self,
target_type: str,
target_uid: str,
row: dict,
title: str,
body: str,
defaults: dict,
totals: dict,
) -> dict:
context = ""
try:
context = build_context(
TYPE_TABLES.get(target_type, target_type),
target_uid,
row,
row.get("user_uid", "") or "",
)
except Exception as exc:
logger.warning("seo meta context failed: %s", exc)
source_text = f"{context}\n\nTITLE: {title}\n\nCONTENT:\n{body}".strip()
raw, usage = gateway_complete(
internal_gateway_key(),
SYSTEM_PROMPT,
source_text,
GENERATION_TIMEOUT_SECONDS,
)
if usage:
for key in totals:
totals[key] += usage.get(key, 0)
parsed = self._parse(raw)
if not parsed:
return {**defaults, "source": "plain"}
clamped = clamp_generated(
parsed.get("seo_title") or defaults["title"],
parsed.get("seo_description") or defaults["description"],
parsed.get("seo_keywords") or defaults["keywords"],
)
return {
"title": clamped["title"] or defaults["title"],
"description": clamped["description"] or defaults["description"],
"keywords": clamped["keywords"] or defaults["keywords"],
"source": "ai",
}
def _parse(self, raw: str) -> dict | None:
if not raw:
return None
text = raw.strip()
if text.startswith("```"):
text = text.strip("`")
if text.lower().startswith("json"):
text = text[4:]
start = text.find("{")
end = text.rfind("}")
if start == -1 or end == -1 or end <= start:
return None
try:
data = json.loads(text[start : end + 1])
except (ValueError, TypeError):
return None
return data if isinstance(data, dict) else None
def _load_row(self, target_type: str, target_uid: str) -> dict | None:
return load_target_row(target_type, target_uid)
def _audit(
self, target_type: str, target_uid: str, source: str, title: str, result: str
) -> None:
from devplacepy.services.audit import record as audit
event = "seo.meta.generate" if result == "success" else "seo.meta.failed"
audit.record_system(
event,
actor_kind="service",
origin="service",
result=result,
target_type=target_type,
target_uid=target_uid,
target_label=title[:120],
metadata={"source": source},
summary=f"seo metadata {result} for {target_type} {target_uid} ({source})",
)
def cleanup(self, job: dict) -> None:
pass
async def run_once(self) -> None:
await super().run_once()
try:
self._backfill()
except Exception as exc:
logger.warning("seo meta backfill failed: %s", exc)
def _backfill(self) -> None:
from devplacepy.database import get_int_setting, get_setting
if get_setting("seo_meta_backfill_enabled", "1") == "0":
return
batch = max(1, get_int_setting("seo_meta_backfill_batch", 5))
for target_type, table in TYPE_TABLES.items():
self._backfill_type(target_type, table, batch)
def _backfill_type(self, target_type: str, table: str, batch: int) -> None:
from devplacepy.database import db
if table not in db.tables:
return
criteria = {"deleted_at": None, "_limit": batch * 4, "order_by": ["-created_at"]}
if target_type == "news":
criteria["status"] = "published"
enqueued = 0
for row in get_table(table).find(**criteria):
if enqueued >= batch:
break
uid = row.get("uid")
if not uid or has_fresh_seo_metadata(target_type, uid):
continue
schedule_seo_meta(target_type, uid)
enqueued += 1
def collect_metrics(self) -> dict:
base = super().collect_metrics()
base["stats"] = base["stats"] + usage_metric_cards(get_seo_usage())
return base
View Git Blame Copy Permalink