+123

apps/import/journal_sources.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + from __future__ import annotations 5 5 + 6 6 + import base64 7 7 + import json 8 8 + import logging 9 9 + import os 10 10 + import secrets 11 11 + from functools import wraps 12 12 + from pathlib import Path 13 13 + 14 14 + from flask import abort, g, request 15 15 + 16 16 + from apps.utils import get_app_storage_path 17 17 + 18 18 + logger = logging.getLogger(__name__) 19 19 + 20 20 + KEY_BYTES = 32 21 21 + STATE_AREAS = ("segments", "entities", "facets", "imports", "config") 22 22 + 23 23 + 24 24 + def is_valid_journal_source_name(name: str) -> bool: 25 25 + return bool(name) and name not in {".", ".."} and "/" not in name and "\\" not in name 26 26 + 27 27 + 28 28 + def generate_key() -> str: 29 29 + return base64.urlsafe_b64encode(secrets.token_bytes(KEY_BYTES)).decode().rstrip("=") 30 30 + 31 31 + 32 32 + def get_journal_sources_dir() -> Path: 33 33 + return get_app_storage_path("import", "journal_sources", ensure_exists=True) 34 34 + 35 35 + 36 36 + def load_journal_source(key: str) -> dict | None: 37 37 + sources_dir = get_journal_sources_dir() 38 38 + for source_path in sources_dir.glob("*.json"): 39 39 + try: 40 40 + with open(source_path, encoding="utf-8") as f: 41 41 + data = json.load(f) 42 42 + if data.get("key") == key: 43 43 + return data 44 44 + except (json.JSONDecodeError, OSError) as e: 45 45 + logger.warning("Failed to load journal source %s: %s", source_path, e) 46 46 + return None 47 47 + 48 48 + 49 49 + def save_journal_source(data: dict) -> bool: 50 50 + name = data.get("name") 51 51 + if not is_valid_journal_source_name(name): 52 52 + return False 53 53 + source_path = get_journal_sources_dir() / f"{name}.json" 54 54 + try: 55 55 + with open(source_path, "w", encoding="utf-8") as f: 56 56 + json.dump(data, f, indent=2) 57 57 + os.chmod(source_path, 0o600) 58 58 + return True 59 59 + except OSError: 60 60 + return False 61 61 + 62 62 + 63 63 + def list_journal_sources() -> list[dict]: 64 64 + sources_dir = get_journal_sources_dir() 65 65 + sources = [] 66 66 + for source_path in sources_dir.glob("*.json"): 67 67 + try: 68 68 + with open(source_path, encoding="utf-8") as f: 69 69 + data = json.load(f) 70 70 + sources.append(data) 71 71 + except (json.JSONDecodeError, OSError): 72 72 + continue 73 73 + sources.sort(key=lambda x: x.get("created_at", 0), reverse=True) 74 74 + return sources 75 75 + 76 76 + 77 77 + def find_journal_source_by_name(name: str) -> dict | None: 78 78 + if not is_valid_journal_source_name(name): 79 79 + return None 80 80 + source_path = get_journal_sources_dir() / f"{name}.json" 81 81 + if not source_path.exists(): 82 82 + return None 83 83 + try: 84 84 + with open(source_path, encoding="utf-8") as f: 85 85 + return json.load(f) 86 86 + except (json.JSONDecodeError, OSError): 87 87 + return None 88 88 + 89 89 + 90 90 + def create_state_directory(journal_root: Path, key_prefix: str) -> Path: 91 91 + state_dir = journal_root / "imports" / key_prefix 92 92 + state_dir.mkdir(parents=True, exist_ok=True) 93 93 + (state_dir / "source.json").write_text("{}", encoding="utf-8") 94 94 + for area in STATE_AREAS: 95 95 + area_dir = state_dir / area 96 96 + area_dir.mkdir(parents=True, exist_ok=True) 97 97 + (area_dir / "state.json").write_text("{}", encoding="utf-8") 98 98 + return state_dir 99 99 + 100 100 + 101 101 + def require_journal_source(f): 102 102 + @wraps(f) 103 103 + def wrapped(*args, **kwargs): 104 104 + auth = request.headers.get("Authorization", "") 105 105 + token = None 106 106 + if auth.startswith("Bearer "): 107 107 + bearer = auth[7:].strip() 108 108 + if bearer: 109 109 + token = bearer 110 110 + 111 111 + if not token: 112 112 + abort(401, description="Missing or invalid authentication") 113 113 + 114 114 + source = load_journal_source(token) 115 115 + if not source: 116 116 + abort(401, description="Invalid API key") 117 117 + if source.get("revoked"): 118 118 + abort(403, description="API key has been revoked") 119 119 + 120 120 + g.journal_source = source 121 121 + return f(*args, **kwargs) 122 122 + 123 123 + return wrapped

+102

apps/import/routes.py

reviewed

··· 12 12 from flask import Blueprint, jsonify, render_template, request 13 13 from werkzeug.utils import secure_filename 14 14 15 15 + from .journal_sources import ( 16 16 + create_state_directory, 17 17 + find_journal_source_by_name, 18 18 + generate_key, 19 19 + is_valid_journal_source_name, 20 20 + list_journal_sources, 21 21 + save_journal_source, 22 22 + ) 23 23 + from apps.utils import log_app_action 15 24 from convey import emit, state 16 25 from media import MEDIA_EXTENSIONS 17 26 from think.detect_created import detect_created ··· 827 836 emit("supervisor", "request", ref=task_id, cmd=cmd) 828 837 829 838 return jsonify({"status": "ok", "task_id": task_id}) 839 839 + 840 840 + 841 841 + @import_bp.route("/api/journal-sources/create", methods=["POST"]) 842 842 + def api_journal_source_create() -> Any: 843 843 + data = request.get_json(force=True) if request.is_json else {} 844 844 + name = data.get("name", "").strip() 845 845 + if not name: 846 846 + return jsonify({"error": "Name is required"}), 400 847 847 + if not is_valid_journal_source_name(name): 848 848 + return jsonify({"error": "Invalid journal source name"}), 400 849 849 + if find_journal_source_by_name(name): 850 850 + return jsonify({"error": f"Journal source '{name}' already exists"}), 409 851 851 + key = generate_key() 852 852 + source_data = { 853 853 + "key": key, 854 854 + "name": name, 855 855 + "created_at": now_ms(), 856 856 + "enabled": True, 857 857 + "revoked": False, 858 858 + "revoked_at": None, 859 859 + "stats": { 860 860 + "segments_received": 0, 861 861 + "entities_received": 0, 862 862 + "facets_received": 0, 863 863 + "imports_received": 0, 864 864 + "config_received": 0, 865 865 + }, 866 866 + } 867 867 + if not save_journal_source(source_data): 868 868 + return jsonify({"error": "Failed to save journal source"}), 500 869 869 + create_state_directory(Path(state.journal_root), key[:8]) 870 870 + log_app_action( 871 871 + app="import", 872 872 + facet=None, 873 873 + action="journal_source_create", 874 874 + params={"name": name, "key_prefix": key[:8]}, 875 875 + ) 876 876 + return jsonify({"key": key, "key_prefix": key[:8], "name": name}) 877 877 + 878 878 + 879 879 + @import_bp.route("/api/journal-sources/list") 880 880 + def api_journal_source_list() -> Any: 881 881 + sources = list_journal_sources() 882 882 + result = [] 883 883 + for s in sources: 884 884 + result.append( 885 885 + { 886 886 + "name": s.get("name", ""), 887 887 + "prefix": s.get("key", "")[:8], 888 888 + "status": "revoked" if s.get("revoked") else "active", 889 889 + "created_at": s.get("created_at"), 890 890 + } 891 891 + ) 892 892 + return jsonify(result) 893 893 + 894 894 + 895 895 + @import_bp.route("/api/journal-sources/<name>/revoke", methods=["POST"]) 896 896 + def api_journal_source_revoke(name: str) -> Any: 897 897 + source = find_journal_source_by_name(name) 898 898 + if not source: 899 899 + return jsonify({"error": f"Journal source '{name}' not found"}), 404 900 900 + if source.get("revoked"): 901 901 + return jsonify({"error": f"Journal source '{name}' is already revoked"}), 409 902 902 + source["revoked"] = True 903 903 + source["revoked_at"] = now_ms() 904 904 + if not save_journal_source(source): 905 905 + return jsonify({"error": "Failed to save journal source"}), 500 906 906 + log_app_action( 907 907 + app="import", 908 908 + facet=None, 909 909 + action="journal_source_revoke", 910 910 + params={"name": name, "key_prefix": source["key"][:8]}, 911 911 + ) 912 912 + return jsonify({"name": name, "prefix": source["key"][:8], "revoked": True}) 913 913 + 914 914 + 915 915 + @import_bp.route("/api/journal-sources/<name>/status") 916 916 + def api_journal_source_status(name: str) -> Any: 917 917 + source = find_journal_source_by_name(name) 918 918 + if not source: 919 919 + return jsonify({"error": f"Journal source '{name}' not found"}), 404 920 920 + key = source.get("key", "") 921 921 + return jsonify( 922 922 + { 923 923 + "name": source.get("name", ""), 924 924 + "prefix": key[:8], 925 925 + "status": "revoked" if source.get("revoked") else "active", 926 926 + "created_at": source.get("created_at"), 927 927 + "revoked": source.get("revoked", False), 928 928 + "revoked_at": source.get("revoked_at"), 929 929 + "stats": source.get("stats", {}), 930 930 + } 931 931 + )

+217

tests/test_journal_sources.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + from __future__ import annotations 5 5 + 6 6 + import json 7 7 + import re 8 8 + import stat 9 9 + from importlib import import_module 10 10 + 11 11 + import pytest 12 12 + from flask import Flask, g, jsonify 13 13 + 14 14 + import convey.state 15 15 + from think.utils import now_ms 16 16 + 17 17 + journal_sources = import_module("apps.import.journal_sources") 18 18 + STATE_AREAS = journal_sources.STATE_AREAS 19 19 + create_state_directory = journal_sources.create_state_directory 20 20 + find_journal_source_by_name = journal_sources.find_journal_source_by_name 21 21 + generate_key = journal_sources.generate_key 22 22 + is_valid_journal_source_name = journal_sources.is_valid_journal_source_name 23 23 + list_journal_sources = journal_sources.list_journal_sources 24 24 + load_journal_source = journal_sources.load_journal_source 25 25 + require_journal_source = journal_sources.require_journal_source 26 26 + save_journal_source = journal_sources.save_journal_source 27 27 + 28 28 + 29 29 + @pytest.fixture 30 30 + def journal_env(tmp_path, monkeypatch): 31 31 + monkeypatch.setattr(convey.state, "journal_root", str(tmp_path), raising=False) 32 32 + (tmp_path / "apps" / "import" / "journal_sources").mkdir(parents=True, exist_ok=True) 33 33 + return tmp_path 34 34 + 35 35 + 36 36 + def _source(name: str, key: str, created_at: int = 0) -> dict: 37 37 + return { 38 38 + "key": key, 39 39 + "name": name, 40 40 + "created_at": created_at, 41 41 + "enabled": True, 42 42 + "revoked": False, 43 43 + "revoked_at": None, 44 44 + "stats": { 45 45 + "segments_received": 0, 46 46 + "entities_received": 0, 47 47 + "facets_received": 0, 48 48 + "imports_received": 0, 49 49 + "config_received": 0, 50 50 + }, 51 51 + } 52 52 + 53 53 + 54 54 + def test_generate_key(): 55 55 + key = generate_key() 56 56 + assert len(key) == 43 57 57 + assert re.fullmatch(r"[A-Za-z0-9_-]{43}", key) 58 58 + 59 59 + 60 60 + def test_create_and_load(journal_env): 61 61 + key = generate_key() 62 62 + source = _source("alpha", key, created_at=123) 63 63 + 64 64 + assert save_journal_source(source) is True 65 65 + 66 66 + loaded = load_journal_source(key) 67 67 + assert loaded == source 68 68 + 69 69 + source_path = journal_env / "apps" / "import" / "journal_sources" / "alpha.json" 70 70 + assert source_path.exists() 71 71 + assert stat.S_IMODE(source_path.stat().st_mode) == 0o600 72 72 + 73 73 + 74 74 + def test_load_wrong_key(journal_env): 75 75 + source = _source("alpha", generate_key(), created_at=123) 76 76 + assert save_journal_source(source) is True 77 77 + 78 78 + assert load_journal_source(generate_key()) is None 79 79 + 80 80 + 81 81 + def test_list_journal_sources(journal_env): 82 82 + first = _source("first", generate_key(), created_at=100) 83 83 + second = _source("second", generate_key(), created_at=300) 84 84 + third = _source("third", generate_key(), created_at=200) 85 85 + 86 86 + assert save_journal_source(first) is True 87 87 + assert save_journal_source(second) is True 88 88 + assert save_journal_source(third) is True 89 89 + 90 90 + assert [source["name"] for source in list_journal_sources()] == [ 91 91 + "second", 92 92 + "third", 93 93 + "first", 94 94 + ] 95 95 + 96 96 + 97 97 + def test_find_by_name(journal_env): 98 98 + source = _source("alpha", generate_key(), created_at=123) 99 99 + assert save_journal_source(source) is True 100 100 + 101 101 + assert find_journal_source_by_name("alpha") == source 102 102 + assert find_journal_source_by_name("nonexistent") is None 103 103 + 104 104 + 105 105 + def test_create_state_directory(journal_env): 106 106 + state_dir = create_state_directory(journal_env, "abcd1234") 107 107 + 108 108 + source_path = state_dir / "source.json" 109 109 + assert source_path.exists() 110 110 + assert json.loads(source_path.read_text(encoding="utf-8")) == {} 111 111 + 112 112 + for area in STATE_AREAS: 113 113 + state_path = state_dir / area / "state.json" 114 114 + assert state_path.exists() 115 115 + assert json.loads(state_path.read_text(encoding="utf-8")) == {} 116 116 + 117 117 + 118 118 + def test_duplicate_name_rejected(journal_env): 119 119 + source = _source("alpha", generate_key(), created_at=123) 120 120 + assert save_journal_source(source) is True 121 121 + 122 122 + assert find_journal_source_by_name("alpha") == source 123 123 + 124 124 + 125 125 + def test_invalid_name_rejected(journal_env): 126 126 + assert is_valid_journal_source_name("../alpha") is False 127 127 + assert save_journal_source(_source("../alpha", generate_key(), created_at=123)) is False 128 128 + assert find_journal_source_by_name("../alpha") is None 129 129 + assert not (journal_env.parent / "alpha.json").exists() 130 130 + 131 131 + 132 132 + def test_revoke_sets_fields(journal_env): 133 133 + key = generate_key() 134 134 + source = _source("alpha", key, created_at=123) 135 135 + assert save_journal_source(source) is True 136 136 + 137 137 + revoked_at = now_ms() 138 138 + source["revoked"] = True 139 139 + source["revoked_at"] = revoked_at 140 140 + assert save_journal_source(source) is True 141 141 + 142 142 + loaded = load_journal_source(key) 143 143 + assert loaded is not None 144 144 + assert loaded["revoked"] is True 145 145 + assert loaded["revoked_at"] == revoked_at 146 146 + 147 147 + 148 148 + def test_auth_decorator_valid_key(journal_env): 149 149 + key = generate_key() 150 150 + source = _source("alpha", key, created_at=123) 151 151 + assert save_journal_source(source) is True 152 152 + 153 153 + app = Flask(__name__) 154 154 + 155 155 + @app.route("/protected") 156 156 + @require_journal_source 157 157 + def protected(): 158 158 + return jsonify({"name": g.journal_source["name"]}) 159 159 + 160 160 + response = app.test_client().get( 161 161 + "/protected", 162 162 + headers={"Authorization": f"Bearer {key}"}, 163 163 + ) 164 164 + 165 165 + assert response.status_code == 200 166 166 + assert response.get_json() == {"name": "alpha"} 167 167 + 168 168 + 169 169 + def test_auth_decorator_missing_key(journal_env): 170 170 + app = Flask(__name__) 171 171 + 172 172 + @app.route("/protected") 173 173 + @require_journal_source 174 174 + def protected(): 175 175 + return jsonify({"name": g.journal_source["name"]}) 176 176 + 177 177 + response = app.test_client().get("/protected") 178 178 + 179 179 + assert response.status_code == 401 180 180 + 181 181 + 182 182 + def test_auth_decorator_invalid_key(journal_env): 183 183 + app = Flask(__name__) 184 184 + 185 185 + @app.route("/protected") 186 186 + @require_journal_source 187 187 + def protected(): 188 188 + return jsonify({"name": g.journal_source["name"]}) 189 189 + 190 190 + response = app.test_client().get( 191 191 + "/protected", 192 192 + headers={"Authorization": "Bearer does-not-exist"}, 193 193 + ) 194 194 + 195 195 + assert response.status_code == 401 196 196 + 197 197 + 198 198 + def test_auth_decorator_revoked_key(journal_env): 199 199 + key = generate_key() 200 200 + source = _source("alpha", key, created_at=123) 201 201 + source["revoked"] = True 202 202 + source["revoked_at"] = now_ms() 203 203 + assert save_journal_source(source) is True 204 204 + 205 205 + app = Flask(__name__) 206 206 + 207 207 + @app.route("/protected") 208 208 + @require_journal_source 209 209 + def protected(): 210 210 + return jsonify({"name": g.journal_source["name"]}) 211 211 + 212 212 + response = app.test_client().get( 213 213 + "/protected", 214 214 + headers={"Authorization": f"Bearer {key}"}, 215 215 + ) 216 216 + 217 217 + assert response.status_code == 403

+16

think/importers/cli.py

reviewed

··· 326 326 if extra and not args.timestamp: 327 327 args.timestamp = extra[0] 328 328 329 329 + # Dispatch journal-source subcommand 330 330 + if args.media == "journal-source": 331 331 + import sys 332 332 + 333 333 + from think.importers.journal_source_cli import main as journal_source_main 334 334 + 335 335 + forwarded_args = sys.argv[1:] 336 336 + if "journal-source" in forwarded_args: 337 337 + idx = forwarded_args.index("journal-source") 338 338 + forwarded_args = forwarded_args[:idx] + forwarded_args[idx + 1 :] 339 339 + else: 340 340 + forwarded_args = extra 341 341 + sys.argv = [sys.argv[0]] + forwarded_args 342 342 + journal_source_main() 343 343 + return 344 344 + 329 345 if args.backends: 330 346 from think.importers.sync import get_syncable_backends 331 347

+308

think/importers/journal_source_cli.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + """CLI for journal source management. 5 5 + 6 6 + Provides commands for creating, listing, revoking, and checking status 7 7 + of journal source registrations for remote journal data import. 8 8 + 9 9 + Usage: 10 10 + sol import journal-source create <name> 11 11 + sol import journal-source list 12 12 + sol import journal-source revoke <name> 13 13 + sol import journal-source status [name] 14 14 + """ 15 15 + 16 16 + from __future__ import annotations 17 17 + 18 18 + import argparse 19 19 + import datetime 20 20 + import json 21 21 + import sys 22 22 + from importlib import import_module 23 23 + from pathlib import Path 24 24 + 25 25 + from think.utils import get_journal, now_ms, setup_cli 26 26 + 27 27 + 28 28 + def _fmt_time(ts: int | None) -> str: 29 29 + if ts is None: 30 30 + return "never" 31 31 + dt = datetime.datetime.fromtimestamp(ts / 1000) 32 32 + return dt.strftime("%Y-%m-%d %H:%M") 33 33 + 34 34 + 35 35 + def _journal_sources(): 36 36 + return import_module("apps.import.journal_sources") 37 37 + 38 38 + 39 39 + def cmd_create(args: argparse.Namespace) -> int: 40 40 + from apps.utils import log_app_action 41 41 + 42 42 + journal_sources = _journal_sources() 43 43 + name = args.name 44 44 + 45 45 + if not journal_sources.is_valid_journal_source_name(name): 46 46 + print(f"Error: invalid journal source name '{name}'", file=sys.stderr) 47 47 + return 1 48 48 + 49 49 + if journal_sources.find_journal_source_by_name(name): 50 50 + print(f"Error: journal source '{name}' already exists", file=sys.stderr) 51 51 + return 1 52 52 + 53 53 + key = journal_sources.generate_key() 54 54 + source_data = { 55 55 + "key": key, 56 56 + "name": name, 57 57 + "created_at": now_ms(), 58 58 + "enabled": True, 59 59 + "revoked": False, 60 60 + "revoked_at": None, 61 61 + "stats": { 62 62 + "segments_received": 0, 63 63 + "entities_received": 0, 64 64 + "facets_received": 0, 65 65 + "imports_received": 0, 66 66 + "config_received": 0, 67 67 + }, 68 68 + } 69 69 + 70 70 + if not journal_sources.save_journal_source(source_data): 71 71 + print("Error: failed to save journal source", file=sys.stderr) 72 72 + return 1 73 73 + 74 74 + journal_sources.create_state_directory(Path(get_journal()), key[:8]) 75 75 + log_app_action( 76 76 + app="import", 77 77 + facet=None, 78 78 + action="journal_source_create", 79 79 + params={"name": name, "key_prefix": key[:8]}, 80 80 + ) 81 81 + 82 82 + if args.json_output: 83 83 + print(json.dumps({"name": name, "key": key, "prefix": key[:8]})) 84 84 + return 0 85 85 + 86 86 + print("Journal source created:") 87 87 + print(f" Name: {name}") 88 88 + print(f" Prefix: {key[:8]}") 89 89 + print(f" api key: {key}") 90 90 + return 0 91 91 + 92 92 + 93 93 + def cmd_list(args: argparse.Namespace) -> int: 94 94 + journal_sources = _journal_sources() 95 95 + sources = journal_sources.list_journal_sources() 96 96 + 97 97 + if args.json_output: 98 98 + print( 99 99 + json.dumps( 100 100 + [ 101 101 + { 102 102 + "name": source.get("name", ""), 103 103 + "prefix": source.get("key", "")[:8], 104 104 + "status": "revoked" if source.get("revoked") else "active", 105 105 + "created_at": source.get("created_at"), 106 106 + } 107 107 + for source in sources 108 108 + ] 109 109 + ) 110 110 + ) 111 111 + return 0 112 112 + 113 113 + if not sources: 114 114 + print("No journal sources registered.") 115 115 + return 0 116 116 + 117 117 + print(f"{'Name':<24} {'Prefix':<10} {'Status':<10} {'Created':<18}") 118 118 + print("-" * 66) 119 119 + for source in sources: 120 120 + print( 121 121 + f"{source.get('name', ''):<24} " 122 122 + f"{source.get('key', '')[:8]:<10} " 123 123 + f"{('revoked' if source.get('revoked') else 'active'):<10} " 124 124 + f"{_fmt_time(source.get('created_at')):<18}" 125 125 + ) 126 126 + return 0 127 127 + 128 128 + 129 129 + def _status_single(name: str, *, json_output: bool = False) -> int: 130 130 + journal_sources = _journal_sources() 131 131 + source = journal_sources.find_journal_source_by_name(name) 132 132 + if not source: 133 133 + print(f"Error: journal source '{name}' not found", file=sys.stderr) 134 134 + return 1 135 135 + 136 136 + key = source.get("key", "") 137 137 + prefix = key[:8] 138 138 + status = "revoked" if source.get("revoked") else "active" 139 139 + state_dir = str(Path(get_journal()) / "imports" / prefix) 140 140 + stats = source.get("stats", {}) 141 141 + 142 142 + if json_output: 143 143 + print( 144 144 + json.dumps( 145 145 + { 146 146 + "name": source.get("name", ""), 147 147 + "prefix": prefix, 148 148 + "status": status, 149 149 + "created_at": source.get("created_at"), 150 150 + "revoked": source.get("revoked", False), 151 151 + "revoked_at": source.get("revoked_at"), 152 152 + "state_dir": state_dir, 153 153 + "stats": stats, 154 154 + } 155 155 + ) 156 156 + ) 157 157 + return 0 158 158 + 159 159 + print(f"Journal source: {source.get('name', '')}") 160 160 + print(f" Prefix: {prefix}") 161 161 + print(f" Status: {status}") 162 162 + print(f" Created: {_fmt_time(source.get('created_at'))}") 163 163 + if source.get("revoked"): 164 164 + print(f" Revoked at: {_fmt_time(source.get('revoked_at'))}") 165 165 + print(f" State dir: {state_dir}") 166 166 + print(" Stats:") 167 167 + print(f" segments: {stats.get('segments_received', 0)}") 168 168 + print(f" entities: {stats.get('entities_received', 0)}") 169 169 + print(f" facets: {stats.get('facets_received', 0)}") 170 170 + print(f" imports: {stats.get('imports_received', 0)}") 171 171 + print(f" config: {stats.get('config_received', 0)}") 172 172 + return 0 173 173 + 174 174 + 175 175 + def _status_all(*, json_output: bool = False) -> int: 176 176 + journal_sources = _journal_sources() 177 177 + sources = journal_sources.list_journal_sources() 178 178 + 179 179 + if json_output: 180 180 + print( 181 181 + json.dumps( 182 182 + [ 183 183 + { 184 184 + "name": source.get("name", ""), 185 185 + "prefix": source.get("key", "")[:8], 186 186 + "status": "revoked" if source.get("revoked") else "active", 187 187 + "created_at": source.get("created_at"), 188 188 + "stats": source.get("stats", {}), 189 189 + "state_dir": str( 190 190 + Path(get_journal()) / "imports" / source.get("key", "")[:8] 191 191 + ), 192 192 + } 193 193 + for source in sources 194 194 + ] 195 195 + ) 196 196 + ) 197 197 + return 0 198 198 + 199 199 + if not sources: 200 200 + print("No journal sources registered.") 201 201 + return 0 202 202 + 203 203 + print( 204 204 + f"{'Name':<20} {'Status':<10} {'Created':<18} " 205 205 + f"{'Seg':>5} {'Ent':>5} {'Fac':>5} {'Imp':>5} {'Cfg':>5}" 206 206 + ) 207 207 + print("-" * 82) 208 208 + for source in sources: 209 209 + stats = source.get("stats", {}) 210 210 + print( 211 211 + f"{source.get('name', ''):<20} " 212 212 + f"{('revoked' if source.get('revoked') else 'active'):<10} " 213 213 + f"{_fmt_time(source.get('created_at')):<18} " 214 214 + f"{stats.get('segments_received', 0):>5} " 215 215 + f"{stats.get('entities_received', 0):>5} " 216 216 + f"{stats.get('facets_received', 0):>5} " 217 217 + f"{stats.get('imports_received', 0):>5} " 218 218 + f"{stats.get('config_received', 0):>5}" 219 219 + ) 220 220 + return 0 221 221 + 222 222 + 223 223 + def cmd_status(args: argparse.Namespace) -> int: 224 224 + if args.name: 225 225 + return _status_single(args.name, json_output=args.json_output) 226 226 + return _status_all(json_output=args.json_output) 227 227 + 228 228 + 229 229 + def cmd_revoke(args: argparse.Namespace) -> int: 230 230 + from apps.utils import log_app_action 231 231 + 232 232 + journal_sources = _journal_sources() 233 233 + source = journal_sources.find_journal_source_by_name(args.name) 234 234 + if not source: 235 235 + print(f"Error: journal source '{args.name}' not found", file=sys.stderr) 236 236 + return 1 237 237 + if source.get("revoked"): 238 238 + print(f"Journal source '{args.name}' is already revoked.", file=sys.stderr) 239 239 + return 1 240 240 + 241 241 + name = source.get("name", "") 242 242 + prefix = source.get("key", "")[:8] 243 243 + source["revoked"] = True 244 244 + source["revoked_at"] = now_ms() 245 245 + 246 246 + if not journal_sources.save_journal_source(source): 247 247 + print("Error: failed to save journal source", file=sys.stderr) 248 248 + return 1 249 249 + 250 250 + log_app_action( 251 251 + app="import", 252 252 + facet=None, 253 253 + action="journal_source_revoke", 254 254 + params={"name": name, "key_prefix": prefix}, 255 255 + ) 256 256 + 257 257 + if args.json_output: 258 258 + print(json.dumps({"name": name, "prefix": prefix, "revoked": True})) 259 259 + return 0 260 260 + 261 261 + print(f"Revoked journal source '{name}' ({prefix})") 262 262 + return 0 263 263 + 264 264 + 265 265 + def main() -> None: 266 266 + parser = argparse.ArgumentParser( 267 267 + prog="sol import journal-source", 268 268 + description="Manage journal source registrations", 269 269 + ) 270 270 + parser.add_argument( 271 271 + "--json", action="store_true", dest="json_output", help="Output as JSON" 272 272 + ) 273 273 + 274 274 + sub = parser.add_subparsers(dest="command") 275 275 + 276 276 + p_create = sub.add_parser("create", help="Create a new journal source") 277 277 + p_create.add_argument("name", help="Name for the journal source") 278 278 + 279 279 + sub.add_parser("list", help="List all registered journal sources") 280 280 + 281 281 + p_status = sub.add_parser("status", help="Show journal source status details") 282 282 + p_status.add_argument( 283 283 + "name", 284 284 + nargs="?", 285 285 + default=None, 286 286 + help="Journal source name (omit for overview)", 287 287 + ) 288 288 + 289 289 + p_revoke = sub.add_parser("revoke", help="Revoke a journal source") 290 290 + p_revoke.add_argument("name", help="Journal source name") 291 291 + 292 292 + args = setup_cli(parser) 293 293 + 294 294 + import convey.state 295 295 + 296 296 + convey.state.journal_root = get_journal() 297 297 + 298 298 + if not args.command: 299 299 + parser.print_help() 300 300 + sys.exit(1) 301 301 + 302 302 + handlers = { 303 303 + "create": cmd_create, 304 304 + "list": cmd_list, 305 305 + "status": cmd_status, 306 306 + "revoke": cmd_revoke, 307 307 + } 308 308 + sys.exit(handlers[args.command](args))