WIS2 Downloader 开发者指南

from shared import get_redis_client

# 获取 Redis 客户端（单例模式）
redis = get_redis_client()

# 像普通 Redis 客户端一样使用
redis.set('key', 'value')
redis.get('key')

from shared import setup_logging

# 配置根日志记录器（在启动时调用一次）
setup_logging()

# 获取模块专用日志记录器
LOGGER = setup_logging(__name__)

LOGGER.info("Message with UTC timestamp")

def run_manager():
    # 1. Create MQTT subscriber
    mqtt_subscriber = Subscriber(**broker_config)

    # 2. Create Redis command listener
    redis_listener = CommandListener(
        subscriber=mqtt_subscriber,
        channel=COMMAND_CHANNEL
    )

    # 3. Start MQTT in separate thread
    mqtt_thread = threading.Thread(target=mqtt_subscriber.start, daemon=True)
    mqtt_thread.start()

    # 4. Start Redis listener (blocks)
    redis_listener.start()

class Subscriber:
    def __init__(self, host, port, uid, pwd, protocol, session):
        # Configure MQTT client with callbacks
        self.client = mqtt.Client(...)
        self.client.on_message = self._on_message

    def _on_message(self, client, userdata, msg):
        # Parse notification, create Celery task
        job = {
            "topic": msg.topic,
            "target": target,
            "filters": filters,
            "payload": json.loads(msg.payload)
        }
        workflow = wis2_download(job)
        workflow.apply_async()

    def subscribe(self, topic, target, filters):
        self.client.subscribe(topic, qos=0)
        self.active_subscriptions[topic] = {...}

    def unsubscribe(self, topic):
        self.client.unsubscribe(topic)
        del self.active_subscriptions[topic]

class CommandListener(threading.Thread):
    def __init__(self, subscriber, channel):
        self.subscriber = subscriber
        self.pubsub = get_redis_client().pubsub()

    def run(self):
        self.pubsub.subscribe(self.channel)
        while not self.stop_event.is_set():
            message = self.pubsub.get_message()
            if message:
                self._process_command(message)

    def _process_command(self, message):
        command = json.loads(message['data'])
        if command['action'] == 'subscribe':
            self.subscriber.subscribe(
                command['topic'],
                command['save_path'],
                command['filters']
            )
        elif command['action'] == 'unsubscribe':
            self.subscriber.unsubscribe(command['topic'])

@app.post('/subscriptions')
def add_subscription():
    data = get_json()
    topic = normalise_topic(data.get('topic'))
    target = normalise_path(data.get('target', ''))
    filters = data.get('filters', {})

    # Publish command to subscriber via Redis PubSub
    command = {
        "action": "subscribe",
        "topic": topic,
        "save_path": target,
        "filters": filters
    }
    publish_command(command, COMMAND_CHANNEL)

    # Persist to Redis for durability
    persist_subscription(topic, target, filters)

    return jsonify({"status": "accepted", ...}), 202

@app.route('/metrics')
def expose_metrics():
    from shared.redis_metrics import generate_prometheus_text, set_gauge
    set_gauge('celery_queue_length', {}, get_queue_length())
    return Response(generate_prometheus_text(), mimetype='text/plain')

from shared import incr_counter
incr_counter('downloads_total', {'cache': hostname, 'media_type': mime_type})

app = Celery('tasks',
             broker=CELERY_BROKER_URL,
             result_backend=CELERY_RESULT_BACKEND)

# 自动发现任务
app.autodiscover_tasks(['task_manager.tasks', 'task_manager.tasks.wis2'])

@app.task(bind=True)
@metrics_collector
def download_from_wis2(self, job):
    result = {...}  # Initialize result dict

    # 1. Extract identifiers
    message_id = job['payload']['id']
    data_id = job['payload']['properties']['data_id']
    filehash = job['payload']['properties']['integrity']['value']

    # 2. Deduplication check
    for key, type in [(message_id, 'by-msg-id'), ...]:
        if get_status(key, type) == STATUS_SUCCESS:
            result['status'] = STATUS_SKIPPED
            return result

    # 3. Acquire lock
    lock_acquired = redis_client.set(lock_key, 1, nx=True, ex=LOCK_EXPIRE)
    if not lock_acquired:
        raise self.retry(countdown=10, max_retries=10)

    # 4. Download file
    response = _pool.request('GET', download_url, ...)

    # 5. Verify hash
    if hash_expected and hash_base64 != hash_expected:
        result['status'] = STATUS_FAILED
        return result

    # 6. Check media type filter
    if not _is_allowed_media_type(file_type, filters):
        result['status'] = STATUS_SKIPPED
        return result

    # 7. Save file
    output_path.write_bytes(data)
    result['status'] = STATUS_SUCCESS
    return result

def wis2_download(args):
    # Chain: download -> decode/ingest
    workflow = download_from_wis2.s(args) | decode_and_ingest.s()
    return workflow

{
  "cache": {
    "children": {
      "a": {
        "children": {
          "wis2": {
            "children": {
              "de-dwd": {
                "children": {
                  ...
                  "synop": {
                    "datasets": [<WCMP2Record>, ...]
                  }
                }
              }
            }
          }
        }
      }
    }
  }
}

from i18n import t

# 简单查找
label = t('nav.dashboard')                      # → 'Dashboard'

# 带插值（使用 str.format）
label = t('subscriptions.folder', path='./')    # → 'Folder: ./'

# modules/ui/i18n/de.py
"""German strings."""

STRINGS: dict[str, str] = {
    'nav.dashboard':        'Dashboard',
    'nav.catalogue':        'Katalogsuche',
    'nav.tree':             'Baumsuche',
    'nav.manual':           'Manuell abonnieren',
    'nav.manage':           'Abonnements verwalten',
    'nav.settings':         'Einstellungen',
    # ... all remaining keys ...
    'subscriptions.folder': 'Ordner: {path}',   # {path} must be preserved
    'settings.records':     '{name}: {count} Einträge',
}

from . import ar, de, en, es, fr, ru, zh        (1)

LANGUAGES: dict[str, str] = {
    'en': 'English',
    'fr': 'Français',
    'es': 'Español',
    'ar': 'العربية',
    'zh': '中文',
    'ru': 'Русский',
    'de': 'Deutsch',                             (2)
}

_STRINGS: dict[str, dict[str, str]] = {
    'en': en.STRINGS,
    'fr': fr.STRINGS,
    'es': es.STRINGS,
    'ar': ar.STRINGS,
    'zh': zh.STRINGS,
    'ru': ru.STRINGS,
    'de': de.STRINGS,                            (3)
}

RTL_LANGUAGES: frozenset[str] = frozenset({'ar', 'he'})

docker-compose restart wis2downloader-ui

# modules/ui/i18n/en.py
'myview.title':       'My New View',
'myview.description': 'Showing results for {topic}.',

# fr.py / es.py / ar.py / zh.py / ru.py
'myview.title':       'My New View',         # TODO: translate
'myview.description': 'Showing results for {topic}.',

from i18n import t

def render(container):
    with container:
        ui.label(t('myview.title')).classes('page-title')
        ui.label(t('myview.description', topic=selected_topic))