gitrusprusandCursor
8fbabd3c11
Replace legacy root backend/frontend with Telegram, Crawl4AI, and VIINA adapters routed by Redis job families. Co-authored-by: Cursor <cursoragent@cursor.com>
33 lines
807 B
Python
33 lines
807 B
Python
"""Redis queue routing for CP adapter families."""
|
|
|
|
from __future__ import annotations
|
|
|
|
# source_type → worker family (separate Redis list + Docker image)
|
|
SOURCE_FAMILY: dict[str, str] = {
|
|
"telegram": "telegram",
|
|
"crawl4ai": "web",
|
|
"viina": "nlp",
|
|
}
|
|
|
|
LEGACY_JOB_QUEUE_KEY = "cp:jobs"
|
|
QUEUE_PREFIX = "cp:jobs"
|
|
|
|
|
|
def family_for_source(source_type: str) -> str:
|
|
try:
|
|
return SOURCE_FAMILY[source_type]
|
|
except KeyError as exc:
|
|
raise ValueError(f"Unknown source_type: {source_type}") from exc
|
|
|
|
|
|
def queue_key_for_source(source_type: str) -> str:
|
|
return f"{QUEUE_PREFIX}:{family_for_source(source_type)}"
|
|
|
|
|
|
def queue_key_for_family(family: str) -> str:
|
|
return f"{QUEUE_PREFIX}:{family}"
|
|
|
|
|
|
def known_source_types() -> list[str]:
|
|
return sorted(SOURCE_FAMILY.keys())
|