MapMil/.cursor/skills/add-parser-adapter/SKILL.md

name, description

name	description
add-parser-adapter	Adds a new CP source_type end-to-end in MapMil: contracts schema and queue family, SourceAdapter implementation, registry, optional Docker worker image, and ParsersView.vue form fields. Use when adding a parser source, adapter, source_type, or new ingest channel to the parsing center.

Add Parser Adapter

End-to-end checklist for a new source_type in ЦП. Work in order; do not skip contracts.

Inputs to clarify

• source_type slug (lowercase, e.g. rss)
• Worker family (telegram | web | nlp) — pick by deps:
  • telegram — Telethon, shared session
  • web — browser/HTTP heavy (Crawl4AI, Playwright)
  • nlp — lightweight HTTP + text (VIINA-style)
• source_config fields for admin UI
• Whether existing worker image can host it or needs new requirements-*.txt service

Checklist

1. Contracts

contracts/sources.py

• Add {Name}SourceConfig(BaseModel) with validators
• Register in CONFIG_MODELS["{source_type}"]

contracts/queues.py

• Add SOURCE_FAMILY["{source_type}"] = "{family}"

No CA code changes needed for routing — enqueue_job uses queue_key_for_source.

2. Adapter

centers/parsing/workers/workers/adapters/{name}.py

Copy pattern from viina.py (text/HTTP) or telegram.py (Telethon):

class MyAdapter:
    source_type = "{source_type}"

    async def run(self, job_id, source_config, *, ctx: WorkerContext) -> tuple[list[dict], str | None]:
        cfg = MySourceConfig.model_validate(source_config or {})
        # fetch / parse → list of IngestEventItem-shaped dicts
        return events, None  # or ([], "error message")

Rules:

• Validate config with Pydantic model from contracts/sources
• Every event: stable source_url, source_type="{source_type}"
• Optional fields: title, description, locality, latitude, longitude, event_date, region, topic, tags, metadata
• Return ([], error) on fatal config errors; partial fetch errors can go in metadata

workers/adapters/registry.py

• Add lazy factory in _factories()
• Factory name key must match source_type

3. Dependencies & Docker

If adapter fits an existing family, only update ENABLED_ADAPTERS in compose (usually not needed — one adapter per service).

If new deps are heavy or conflict:

1. centers/parsing/workers/requirements-{family}.txt (or new file)
2. New or updated service in docker-compose.yml:

cp-workers-{family}:
  build:
    context: .
    dockerfile: centers/parsing/workers/Dockerfile
    args:
      REQUIREMENTS_FILE: requirements-{family}.txt
      INSTALL_PLAYWRIGHT: "0"  # "1" only for browser adapters
  environment:
    ENABLED_ADAPTERS: {source_type}
    WORKER_FAMILIES: {family}
    CA_API_URL: http://ca-api:8000
    REDIS_URL: redis://redis:6379/0
    INTERNAL_TOKEN: dev-internal-token
  depends_on: [redis, ca-api]

4. Admin UI

centers/analytics/frontend/src/views/ParsersView.vue

• Add option to source type <select>
• Extend form state + buildSourceConfig() for new fields
• Add create/edit <template> blocks (mirror crawl4ai or viina)
• Extend validation in submit handler

TypeScript unions: update "telegram" | "crawl4ai" | "viina" wherever used.

No CA router changes if source_type is registered in CONFIG_MODELS — _validated_source_config picks it up automatically.

5. Documentation

• Add row to adapter table in centers/parsing/ARCHITECTURE.md
• One-line mention in README.md if user-facing

6. Verify

docker compose config
docker compose up --build cp-workers-{family}   # or relevant service
curl -X POST http://localhost:8080/admin/jobs \
  -H 'Content-Type: application/json' \
  -d '{"source_type":"{source_type}","source_config":{...},"interval_seconds":3600}'

Check CA logs for ingest and /admin/events for new rows.

Reference files

Concern	File
Config schema	contracts/sources.py
Queue routing	contracts/queues.py
Ingest shape	contracts/ingest.py
Adapter protocol	workers/adapters/base.py
Simple adapter	workers/adapters/viina.py
LLM adapter	workers/adapters/crawl4ai_adapter.py
Telegram + listener	workers/adapters/telegram.py
Enqueue	centers/analytics/api/app/services/jobs.py
Admin validation	centers/analytics/api/app/routers/admin.py

Anti-patterns

• Implementing adapter before CONFIG_MODELS entry
• Hardcoding Redis queue names — use queue_key_for_source
• Skipping source_url stability (breaks dedup)
• Importing heavy deps at module top in registry.py — use lazy factories
• Editing Telegram listener when adding unrelated sources