# retoor import logging from devplacepy.database import ( SEO_META_TYPES, get_table, has_fresh_seo_metadata, mark_seo_metadata_stale, ) from devplacepy.seo_meta_text import plain_seo_defaults from devplacepy.services.jobs import queue logger = logging.getLogger(__name__) TABLE_TO_TYPE = { "posts": "post", "projects": "project", "gists": "gist", "news": "news", "quizzes": "quiz", } TYPE_TABLES = { "post": "posts", "project": "projects", "gist": "gists", "news": "news", "quiz": "quizzes", } TITLE_FIELD = { "post": "title", "project": "title", "gist": "title", "news": "title", "quiz": "title", "issue": "enhanced_title", } BODY_FIELD = { "post": "content", "project": "description", "gist": "description", "news": "description", "quiz": "description", "issue": "original_description", } def target_type_for_table(table: str) -> str: return TABLE_TO_TYPE.get(table, "") def load_target_row(target_type: str, target_uid: str) -> dict | None: table = TYPE_TABLES.get(target_type) if table: row = get_table(table).find_one(uid=target_uid, deleted_at=None) return dict(row) if row else None if target_type == "issue": from devplacepy.services.gitea import store try: row = store.get_ticket(int(target_uid)) except (ValueError, TypeError): return None return dict(row) if row else None return None def defaults_for_target(target_type: str, target_uid: str) -> dict: row = load_target_row(target_type, target_uid) if not row: return plain_seo_defaults("", "") title = row.get(TITLE_FIELD.get(target_type, "title")) or "" body = row.get(BODY_FIELD.get(target_type, "content")) or "" return plain_seo_defaults(title, body) def schedule_seo_meta(target_type: str, target_uid: str, regenerate: bool = False) -> None: if target_type not in SEO_META_TYPES or not target_uid: return target_uid = str(target_uid) if regenerate: mark_seo_metadata_stale(target_type, target_uid) elif has_fresh_seo_metadata(target_type, target_uid): return if _has_pending_job(target_type, target_uid): return queue.enqueue( "seo_meta", {"target_type": target_type, "target_uid": target_uid}, "system", "seo_meta", preferred_name=f"{target_type}:{target_uid[:12]}", ) def schedule_seo_meta_for_table(table: str, target_uid: str, regenerate: bool = False) -> None: target_type = target_type_for_table(table) if target_type: schedule_seo_meta(target_type, target_uid, regenerate=regenerate) def _has_pending_job(target_type: str, target_uid: str) -> bool: for job in queue.list_jobs(kind="seo_meta", status=queue.PENDING): payload = job.get("payload") or {} if ( payload.get("target_type") == target_type and str(payload.get("target_uid")) == target_uid ): return True for job in queue.list_jobs(kind="seo_meta", status=queue.RUNNING): payload = job.get("payload") or {} if ( payload.get("target_type") == target_type and str(payload.get("target_uid")) == target_uid ): return True return False